You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/10/12 03:24:42 UTC
(skywalking) branch master updated: Add configurable HTTP timeout for alarm hooks (#14128)
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/master by this push:
new 653eef50e3 Add configurable HTTP timeout for alarm hooks (#14128)
653eef50e3 is described below
commit 653eef50e34ba0cf1838937b71d567dc73276ea3
Author: songzhendong <xi...@qq.com>
AuthorDate: Mon Oct 12 11:24:35 2026 +0800
Add configurable HTTP timeout for alarm hooks (#14128)
Every HTTP-based alarm hook (Webhook, Slack, WeChat, DingTalk, Feishu,
WeLink, PagerDuty, Discord) now bounds each HTTP request it sends by one
deadline, alarm.default.httpTimeout (SW_ALARM_HTTP_TIMEOUT, default 12
seconds) in application.yml, covering connecting, waiting for the
response and reading the response body. A value <= 0 falls back to 12.
The gRPC hook is unchanged.
The previous fixed 12-second request timeout stopped once the response
headers arrived, so a hook endpoint that stalled while sending the body
could hold the single alarm delivery thread indefinitely.
JdkHttpClientUtils.sendStringWithTimeout in library-util gives the
headers and the body one shared budget, and on timeout or interruption
cancels the body subscription, since on Java 11 cancelling the pending
request alone leaves a stalled connection open. It requires the request
to carry its own timeout, the only thing that aborts a request still
waiting for the response headers.
AlarmModuleProvider builds one HttpClient shared by every hook and passes
it with the timeout through NotifyHandler to each hook's constructor.
The alarm and configuration docs describe the setting.
---
docs/en/changes/changes.md | 1 +
docs/en/setup/backend/backend-alarm.md | 16 +
docs/en/setup/backend/configuration-vocabulary.md | 1 +
.../core/alarm/provider/AlarmModuleConfig.java | 39 ++
.../core/alarm/provider/AlarmModuleProvider.java | 35 +-
.../server/core/alarm/provider/NotifyHandler.java | 33 +-
.../provider/dingtalk/DingtalkHookCallback.java | 10 +-
.../provider/discord/DiscordHookCallback.java | 10 +-
.../alarm/provider/feishu/FeishuHookCallback.java | 10 +-
.../provider/pagerduty/PagerDutyHookCallback.java | 10 +-
.../alarm/provider/slack/SlackhookCallback.java | 10 +-
.../alarm/provider/webhook/WebhookCallback.java | 10 +-
.../alarm/provider/wechat/WechatHookCallback.java | 10 +-
.../alarm/provider/welink/WeLinkHookCallback.java | 10 +-
.../alarm/provider/AlarmModuleProviderTest.java | 42 ++
.../core/alarm/provider/NotifyHandlerTest.java | 5 +-
.../dingtalk/DingtalkHookCallbackTest.java | 6 +-
.../provider/feishu/FeishuHookCallbackTest.java | 8 +-
.../pagerduty/PagerDutyHookCallbackTest.java | 6 +-
.../provider/slack/SlackHookCallbackTest.java | 4 +-
.../provider/webhook/WebhookCallbackTest.java | 4 +-
.../provider/wechat/WechatHookCallbackTest.java | 4 +-
.../provider/welink/WeLinkHookCallbackTest.java | 4 +-
.../oap/server/core/alarm/HttpAlarmCallback.java | 29 +-
.../server/core/alarm/HttpAlarmCallbackTest.java | 54 ++-
.../server/library/util/JdkHttpClientUtils.java | 184 +++++++++
.../library/util/JdkHttpClientUtilsTest.java | 426 +++++++++++++++++++++
.../src/main/resources/application.yml | 3 +
.../src/main/resources/application.yml | 3 +
test/e2e-v2/cases/storage/expected/config-dump.yml | 1 +
30 files changed, 939 insertions(+), 49 deletions(-)
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index e6d421ba59..5f5e732705 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -35,6 +35,7 @@
* Fix the Elasticsearch `BulkProcessor` treating a bulk write as fully successful whenever the overall HTTP status was 200, even when Elasticsearch rejected individual items in the same response, for example a 429 `cluster_block_exception` when an index is switched to `read_only_allow_delete` by the flood-stage disk watermark. The response body's `errors` flag and per-item `status`/`error` are now decoded and checked on every 200 response; a rejected request's future now completes except [...]
* AI agent conversations adopt AI Sessionizer `6691c36`, and withhold what the runtime sent the model. The Sessionizer names the records that carry the system prompt and the tool schemas with the flags `system_prompt` and `tool_schemas`; the `hide` setting (`SW_AI_AGENT_CONVERSATION_HIDE`) names what this OAP withholds from every reader; the routes take no parameter for it. Withholding happens as a conversation is served, and only then: the storage holds every file whole, and both routes [...]
* Fix endpoints registered from logs missing from `findEndpoint` (the endpoint list) when it is queried with a time range. The query filters on an endpoint's `last_ping`, which endpoints registered from logs, from virtual MQ spans, from cross-thread segment references and from Cilium Hubble never set, so they kept `last_ping` at `0`, and an endpoint also seen in traces was reset to `0` whenever one of those sources wrote it last. Every endpoint source now sets `last_ping`, and an update [...]
+* Add `alarm.default.httpTimeout` (`SW_ALARM_HTTP_TIMEOUT`, default 12 seconds) in `application.yml`, the deadline of each HTTP request an HTTP-based alarm hook (Webhook, Slack, WeChat, DingTalk, Feishu, WeLink, PagerDuty, Discord) sends, covering connecting, waiting for the response and reading the response body; the gRPC hook is unchanged. The previous fixed 12-second request timeout stopped once the response headers arrived, so a hook endpoint that stalled while sending the body could [...]
#### UI
* Add a Virtual GenAI evaluation-record page and evaluation-score chart in Horizon UI, so operators can inspect evaluation result, level, reason, judge model, timestamp, trace linkage, and the `gen_ai_model_evaluation_score_ppm` trend for evaluated records.
diff --git a/docs/en/setup/backend/backend-alarm.md b/docs/en/setup/backend/backend-alarm.md
index d1abcf638d..57400caa9a 100644
--- a/docs/en/setup/backend/backend-alarm.md
+++ b/docs/en/setup/backend/backend-alarm.md
@@ -509,6 +509,22 @@ discord:
username: robot
```
+### HTTP delivery timeout
+Every HTTP-based hook (Webhook, Slack, WeChat, DingTalk, Feishu, WeLink, PagerDuty and Discord) bounds each HTTP request
+it sends by one deadline, covering connecting, waiting for the response and reading the response body. A hook that sends
+more than one request for an alarm gets the full deadline for each of them, for example WeLink fetching its access token
+before sending the message. The gRPC hook does not use it.
+
+The deadline is set in `config/application.yml`, not in `alarm-settings.yml`, so it does not change dynamically:
+
+```yaml
+alarm:
+ default:
+ httpTimeout: ${SW_ALARM_HTTP_TIMEOUT:12} # seconds
+```
+
+The default is `12` seconds. Values `<= 0` fall back to `12`.
+
## Update the settings dynamically
Since 6.5.0, the alerting settings can be updated dynamically at runtime by [Dynamic Configuration](dynamic-config.md),
which will override the settings in `alarm-settings.yml`.
diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md
index 2be8492341..24fea43a2c 100644
--- a/docs/en/setup/backend/configuration-vocabulary.md
+++ b/docs/en/setup/backend/configuration-vocabulary.md
@@ -315,6 +315,7 @@ It divided into several modules, each of which has its own settings. The followi
| - | - | skywalkingTracesListResultTags | Span tag keys included in the trace list result for the SkyWalking native data source. These tags are used for trace grouping and shown in the trace list panel. Multiple values are separated by commas. [...]
| - | - | otlpTracesListResultTags | Span attribute keys included in the trace list result of the OTLP data source, used by Grafana for grouping. | SW_TRACEQL_OTLP_TRACES_LIST_RESULT_TAGS | http.request.method,http.response.status_code |
| alarm | default | - | Read [alarm doc](backend-alarm.md) for more details. [...]
+| - | - | httpTimeout | Deadline (seconds) of each HTTP request an HTTP-based alarm hook (every hook except gRPC) sends, covering connecting, waiting for the response and reading the response body. Values `<= 0` fall back to `12`. [...]
| telemetry | - | - | Read [telemetry doc](backend-telemetry.md) for more details. [...]
| - | none | - | No op implementation. [...]
| - | prometheus | host | Binding host for Prometheus server fetching data. [...]
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleConfig.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleConfig.java
new file mode 100644
index 0000000000..cea5c12a17
--- /dev/null
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleConfig.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.skywalking.oap.server.core.alarm.provider;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.skywalking.oap.server.library.module.ModuleConfig;
+
+/**
+ * Settings under {@code alarm.default} in {@code application.yml}.
+ */
+@Getter
+@Setter
+public class AlarmModuleConfig extends ModuleConfig {
+ public static final long DEFAULT_HTTP_TIMEOUT_SECONDS = 12;
+
+ /**
+ * Deadline (seconds) of each HTTP request an HTTP-based hook makes, covering connect, waiting for the
+ * response and reading the response body. Values {@code <= 0} fall back to
+ * {@link #DEFAULT_HTTP_TIMEOUT_SECONDS}. The gRPC hook does not use it.
+ */
+ private long httpTimeout = DEFAULT_HTTP_TIMEOUT_SECONDS;
+}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProvider.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProvider.java
index ed4513a916..a5f0f0cb93 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProvider.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProvider.java
@@ -20,6 +20,8 @@ package org.apache.skywalking.oap.server.core.alarm.provider;
import java.io.FileNotFoundException;
import java.io.Reader;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
import lombok.Getter;
import org.apache.skywalking.oap.server.configuration.api.ConfigurationModule;
import org.apache.skywalking.oap.server.configuration.api.DynamicConfigurationService;
@@ -29,7 +31,9 @@ import org.apache.skywalking.oap.server.core.alarm.AlarmModule;
import org.apache.skywalking.oap.server.core.alarm.AlarmRulesWatcherService;
import org.apache.skywalking.oap.server.core.alarm.AlarmStandardPersistence;
import org.apache.skywalking.oap.server.core.alarm.AlarmStatusWatcherService;
+import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;
import org.apache.skywalking.oap.server.core.alarm.MetricsNotify;
+import org.apache.skywalking.oap.server.library.module.ModuleConfig;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
@@ -37,10 +41,12 @@ import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedExcepti
import org.apache.skywalking.oap.server.library.util.ResourceUtils;
public class AlarmModuleProvider extends ModuleProvider {
+ private static final long MAX_HTTP_TIMEOUT_SECONDS = TimeUnit.NANOSECONDS.toSeconds(Long.MAX_VALUE);
private NotifyHandler notifyHandler;
@Getter
private AlarmRulesWatcher alarmRulesWatcher;
+ private AlarmModuleConfig moduleConfig = new AlarmModuleConfig();
@Override
public String name() {
@@ -53,14 +59,26 @@ public class AlarmModuleProvider extends ModuleProvider {
}
@Override
- public ConfigCreator newConfigCreator() {
- return null;
+ public ConfigCreator<? extends ModuleConfig> newConfigCreator() {
+ return new ConfigCreator<AlarmModuleConfig>() {
+ @Override
+ public Class<AlarmModuleConfig> type() {
+ return AlarmModuleConfig.class;
+ }
+
+ @Override
+ public void onInitialized(final AlarmModuleConfig initialized) {
+ moduleConfig = initialized;
+ }
+ };
}
@Override
public void prepare() throws ServiceNotProvidedException, ModuleStartException {
alarmRulesWatcher = new AlarmRulesWatcher(new Rules(), this, getManager());
- notifyHandler = new NotifyHandler(alarmRulesWatcher, getManager());
+ final Duration httpTimeout = httpTimeout(moduleConfig.getHttpTimeout());
+ notifyHandler = new NotifyHandler(
+ alarmRulesWatcher, getManager(), HttpAlarmCallback.newHttpClient(httpTimeout), httpTimeout);
this.registerServiceImplementation(MetricsNotify.class, notifyHandler);
this.registerServiceImplementation(AlarmRulesWatcherService.class, alarmRulesWatcher);
this.registerServiceImplementation(AlarmStatusWatcherService.class, new AlarmStatusWatcher(getManager()));
@@ -91,6 +109,17 @@ public class AlarmModuleProvider extends ModuleProvider {
notifyHandler.init(new AlarmStandardPersistence(getManager()));
}
+ /**
+ * The cap keeps the JDK's own timer arithmetic from overflowing: {@code Long.MAX_VALUE} seconds is accepted by
+ * {@code HttpRequest.Builder#timeout} and then fails every send with an {@link ArithmeticException}.
+ */
+ static Duration httpTimeout(final long seconds) {
+ if (seconds <= 0) {
+ return Duration.ofSeconds(AlarmModuleConfig.DEFAULT_HTTP_TIMEOUT_SECONDS);
+ }
+ return Duration.ofSeconds(Math.min(seconds, MAX_HTTP_TIMEOUT_SECONDS));
+ }
+
@Override
public String[] requiredModules() {
return new String[] {
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandler.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandler.java
index 0d3473734a..55da6c5e4a 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandler.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandler.java
@@ -18,6 +18,8 @@
package org.apache.skywalking.oap.server.core.alarm.provider;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -58,12 +60,17 @@ public class NotifyHandler implements MetricsNotify {
private final AlarmCore core;
private final AlarmRulesWatcher alarmRulesWatcher;
private final ModuleManager manager;
+ private final HttpClient httpClient;
+ private final Duration httpTimeout;
private MetadataQueryService metadataQueryService;
- public NotifyHandler(AlarmRulesWatcher alarmRulesWatcher, ModuleManager manager) {
+ public NotifyHandler(final AlarmRulesWatcher alarmRulesWatcher, final ModuleManager manager,
+ final HttpClient httpClient, final Duration httpTimeout) {
this.alarmRulesWatcher = alarmRulesWatcher;
core = new AlarmCore(alarmRulesWatcher);
this.manager = manager;
+ this.httpClient = httpClient;
+ this.httpTimeout = httpTimeout;
}
private MetadataQueryService metadata() {
@@ -234,16 +241,20 @@ public class NotifyHandler implements MetricsNotify {
}
public void init(AlarmCallback... callbacks) {
- List<AlarmCallback> allCallbacks = new ArrayList<>(Arrays.asList(callbacks));
- allCallbacks.add(new WebhookCallback(alarmRulesWatcher));
+ core.start(buildCallbacks(callbacks));
+ }
+
+ List<AlarmCallback> buildCallbacks(final AlarmCallback... callbacks) {
+ final List<AlarmCallback> allCallbacks = new ArrayList<>(Arrays.asList(callbacks));
+ allCallbacks.add(new WebhookCallback(alarmRulesWatcher, httpClient, httpTimeout));
allCallbacks.add(new GRPCCallback(alarmRulesWatcher));
- allCallbacks.add(new SlackhookCallback(alarmRulesWatcher));
- allCallbacks.add(new WechatHookCallback(alarmRulesWatcher));
- allCallbacks.add(new DingtalkHookCallback(alarmRulesWatcher));
- allCallbacks.add(new FeishuHookCallback(alarmRulesWatcher));
- allCallbacks.add(new WeLinkHookCallback(alarmRulesWatcher));
- allCallbacks.add(new PagerDutyHookCallback(alarmRulesWatcher));
- allCallbacks.add(new DiscordHookCallback(alarmRulesWatcher));
- core.start(allCallbacks);
+ allCallbacks.add(new SlackhookCallback(alarmRulesWatcher, httpClient, httpTimeout));
+ allCallbacks.add(new WechatHookCallback(alarmRulesWatcher, httpClient, httpTimeout));
+ allCallbacks.add(new DingtalkHookCallback(alarmRulesWatcher, httpClient, httpTimeout));
+ allCallbacks.add(new FeishuHookCallback(alarmRulesWatcher, httpClient, httpTimeout));
+ allCallbacks.add(new WeLinkHookCallback(alarmRulesWatcher, httpClient, httpTimeout));
+ allCallbacks.add(new PagerDutyHookCallback(alarmRulesWatcher, httpClient, httpTimeout));
+ allCallbacks.add(new DiscordHookCallback(alarmRulesWatcher, httpClient, httpTimeout));
+ return allCallbacks;
}
}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallback.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallback.java
index 62957b347a..bbbc83b1b5 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallback.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallback.java
@@ -18,7 +18,6 @@
package org.apache.skywalking.oap.server.core.alarm.provider.dingtalk;
-import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;
@@ -31,9 +30,11 @@ import javax.crypto.spec.SecretKeySpec;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URLEncoder;
+import java.net.http.HttpClient;
import java.nio.charset.StandardCharsets;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
import java.util.Base64;
import java.util.List;
import java.util.Map;
@@ -42,10 +43,15 @@ import java.util.Map;
* Use SkyWalking alarm dingtalk webhook API.
*/
@Slf4j
-@RequiredArgsConstructor
public class DingtalkHookCallback extends HttpAlarmCallback {
private final AlarmRulesWatcher alarmRulesWatcher;
+ public DingtalkHookCallback(final AlarmRulesWatcher alarmRulesWatcher, final HttpClient httpClient,
+ final Duration httpTimeout) {
+ super(httpClient, httpTimeout);
+ this.alarmRulesWatcher = alarmRulesWatcher;
+ }
+
protected void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) throws Exception {
Map<String, DingtalkSettings> settingsMap = alarmRulesWatcher.getDingtalkSettings();
if (settingsMap == null || settingsMap.isEmpty()) {
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/discord/DiscordHookCallback.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/discord/DiscordHookCallback.java
index e66c1f55e4..e2e1832a62 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/discord/DiscordHookCallback.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/discord/DiscordHookCallback.java
@@ -19,7 +19,6 @@
package org.apache.skywalking.oap.server.core.alarm.provider.discord;
import com.google.gson.JsonObject;
-import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;
@@ -27,6 +26,8 @@ import org.apache.skywalking.oap.server.core.alarm.provider.AlarmRulesWatcher;
import java.io.IOException;
import java.net.URI;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
@@ -37,10 +38,15 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
* Use SkyWalking alarm Discord webhook API.
*/
@Slf4j
-@RequiredArgsConstructor
public class DiscordHookCallback extends HttpAlarmCallback {
private final AlarmRulesWatcher alarmRulesWatcher;
+ public DiscordHookCallback(final AlarmRulesWatcher alarmRulesWatcher, final HttpClient httpClient,
+ final Duration httpTimeout) {
+ super(httpClient, httpTimeout);
+ this.alarmRulesWatcher = alarmRulesWatcher;
+ }
+
public void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) throws Exception {
Map<String, DiscordSettings> settingsMap = alarmRulesWatcher.getDiscordSettings();
if (settingsMap == null || settingsMap.isEmpty()) {
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/feishu/FeishuHookCallback.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/feishu/FeishuHookCallback.java
index 3973eb7eb5..86d774bf2e 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/feishu/FeishuHookCallback.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/feishu/FeishuHookCallback.java
@@ -20,7 +20,6 @@ package org.apache.skywalking.oap.server.core.alarm.provider.feishu;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
-import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;
@@ -31,8 +30,10 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.net.URI;
+import java.net.http.HttpClient;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Base64;
import java.util.HashMap;
@@ -44,10 +45,15 @@ import java.util.stream.Collectors;
* Use SkyWalking alarm feishu webhook API.
*/
@Slf4j
-@RequiredArgsConstructor
public class FeishuHookCallback extends HttpAlarmCallback {
private final AlarmRulesWatcher alarmRulesWatcher;
+ public FeishuHookCallback(final AlarmRulesWatcher alarmRulesWatcher, final HttpClient httpClient,
+ final Duration httpTimeout) {
+ super(httpClient, httpTimeout);
+ this.alarmRulesWatcher = alarmRulesWatcher;
+ }
+
protected void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) throws Exception {
Map<String, FeishuSettings> settingsMap = alarmRulesWatcher.getFeishuSettings();
if (settingsMap == null || settingsMap.isEmpty()) {
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/pagerduty/PagerDutyHookCallback.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/pagerduty/PagerDutyHookCallback.java
index 2df573fb60..c64a8f5f57 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/pagerduty/PagerDutyHookCallback.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/pagerduty/PagerDutyHookCallback.java
@@ -21,13 +21,14 @@ package org.apache.skywalking.oap.server.core.alarm.provider.pagerduty;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonPrimitive;
-import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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 java.net.URI;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -36,12 +37,17 @@ import org.apache.skywalking.oap.server.library.util.CollectionUtils;
import org.apache.skywalking.oap.server.library.util.StringUtil;
@Slf4j
-@RequiredArgsConstructor
public class PagerDutyHookCallback extends HttpAlarmCallback {
private static final Gson GSON = new Gson();
private final AlarmRulesWatcher alarmRulesWatcher;
+ public PagerDutyHookCallback(final AlarmRulesWatcher alarmRulesWatcher, final HttpClient httpClient,
+ final Duration httpTimeout) {
+ super(httpClient, httpTimeout);
+ this.alarmRulesWatcher = alarmRulesWatcher;
+ }
+
@Override
protected void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) throws Exception {
Map<String, PagerDutySettings> settingsMap = alarmRulesWatcher.getPagerDutySettings();
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/slack/SlackhookCallback.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/slack/SlackhookCallback.java
index 27dab018a2..32a1265b16 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/slack/SlackhookCallback.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/slack/SlackhookCallback.java
@@ -21,13 +21,14 @@ package org.apache.skywalking.oap.server.core.alarm.provider.slack;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
-import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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 java.net.URI;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
@@ -38,12 +39,17 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
* Use SkyWalking alarm slack webhook API calls a remote endpoints.
*/
@Slf4j
-@RequiredArgsConstructor
public class SlackhookCallback extends HttpAlarmCallback {
private static final Gson GSON = new Gson();
private final AlarmRulesWatcher alarmRulesWatcher;
+ public SlackhookCallback(final AlarmRulesWatcher alarmRulesWatcher, final HttpClient httpClient,
+ final Duration httpTimeout) {
+ super(httpClient, httpTimeout);
+ this.alarmRulesWatcher = alarmRulesWatcher;
+ }
+
@Override
public void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) throws Exception {
Map<String, SlackSettings> settingsMap = alarmRulesWatcher.getSlackSettings();
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/webhook/WebhookCallback.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/webhook/WebhookCallback.java
index 0217ddb469..b95b305210 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/webhook/WebhookCallback.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/webhook/WebhookCallback.java
@@ -19,12 +19,13 @@
package org.apache.skywalking.oap.server.core.alarm.provider.webhook;
import com.google.gson.Gson;
-import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;
import java.net.URI;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
@@ -35,11 +36,16 @@ import org.apache.skywalking.oap.server.library.util.CollectionUtils;
* Use SkyWalking alarm webhook API calls a remote endpoints.
*/
@Slf4j
-@RequiredArgsConstructor
public class WebhookCallback extends HttpAlarmCallback {
private final AlarmRulesWatcher alarmRulesWatcher;
private final Gson gson = new Gson();
+ public WebhookCallback(final AlarmRulesWatcher alarmRulesWatcher, final HttpClient httpClient,
+ final Duration httpTimeout) {
+ super(httpClient, httpTimeout);
+ this.alarmRulesWatcher = alarmRulesWatcher;
+ }
+
@Override
public void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) throws Exception {
Map<String, WebhookSettings> settingsMap = alarmRulesWatcher.getWebHooks();
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/wechat/WechatHookCallback.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/wechat/WechatHookCallback.java
index af4a6391cb..f4ed37f38d 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/wechat/WechatHookCallback.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/wechat/WechatHookCallback.java
@@ -18,13 +18,14 @@
package org.apache.skywalking.oap.server.core.alarm.provider.wechat;
-import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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 java.net.URI;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
@@ -35,10 +36,15 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
* Use SkyWalking alarm wechat webhook API.
*/
@Slf4j
-@RequiredArgsConstructor
public class WechatHookCallback extends HttpAlarmCallback {
private final AlarmRulesWatcher alarmRulesWatcher;
+ public WechatHookCallback(final AlarmRulesWatcher alarmRulesWatcher, final HttpClient httpClient,
+ final Duration httpTimeout) {
+ super(httpClient, httpTimeout);
+ this.alarmRulesWatcher = alarmRulesWatcher;
+ }
+
@Override
public void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) throws Exception {
Map<String, WechatSettings> settingsMap = alarmRulesWatcher.getWechatSettings();
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/welink/WeLinkHookCallback.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/welink/WeLinkHookCallback.java
index 8a71eccdbf..ebefb335d3 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/welink/WeLinkHookCallback.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/welink/WeLinkHookCallback.java
@@ -25,7 +25,6 @@ import com.google.gson.JsonObject;
import java.util.Map;
-import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;
@@ -33,6 +32,8 @@ import org.apache.skywalking.oap.server.core.alarm.provider.AlarmRulesWatcher;
import java.io.IOException;
import java.net.URI;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -47,10 +48,15 @@ import org.apache.skywalking.oap.server.library.util.StringUtil;
* Use SkyWalking alarm WeLink webhook API.
*/
@Slf4j
-@RequiredArgsConstructor
public class WeLinkHookCallback extends HttpAlarmCallback {
private final AlarmRulesWatcher alarmRulesWatcher;
+ public WeLinkHookCallback(final AlarmRulesWatcher alarmRulesWatcher, final HttpClient httpClient,
+ final Duration httpTimeout) {
+ super(httpClient, httpTimeout);
+ this.alarmRulesWatcher = alarmRulesWatcher;
+ }
+
/**
* Send alarm message if the settings not empty
*/
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProviderTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProviderTest.java
index 6eb3f3628c..9d2db56e78 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProviderTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProviderTest.java
@@ -18,11 +18,17 @@
package org.apache.skywalking.oap.server.core.alarm.provider;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.Iterator;
+import java.util.List;
import java.util.ServiceLoader;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import org.apache.skywalking.oap.server.configuration.api.ConfigurationModule;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.alarm.AlarmModule;
+import org.apache.skywalking.oap.server.core.alarm.HttpAlarmCallback;
import org.apache.skywalking.oap.server.core.query.enumeration.Scope;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
@@ -33,6 +39,7 @@ import org.apache.skywalking.oap.server.testing.util.ReflectUtil;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -85,4 +92,39 @@ public class AlarmModuleProviderTest {
ConfigurationModule.NAME
}, modules);
}
+
+ @Test
+ public void httpTimeoutFallsBackAndIsCapped() {
+ assertEquals(Duration.ofSeconds(5), AlarmModuleProvider.httpTimeout(5));
+ assertEquals(Duration.ofSeconds(12), AlarmModuleProvider.httpTimeout(0));
+ assertEquals(Duration.ofSeconds(12), AlarmModuleProvider.httpTimeout(-3));
+ assertEquals(
+ Duration.ofSeconds(TimeUnit.NANOSECONDS.toSeconds(Long.MAX_VALUE)),
+ AlarmModuleProvider.httpTimeout(Long.MAX_VALUE)
+ );
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void configuredHttpTimeoutReachesTheHttpHooks() throws Exception {
+ final AlarmModuleConfig config = new AlarmModuleConfig();
+ config.setHttpTimeout(3);
+ ((ModuleProvider.ConfigCreator<AlarmModuleConfig>) moduleProvider.newConfigCreator()).onInitialized(config);
+ moduleProvider.prepare();
+
+ final NotifyHandler handler = ReflectUtil.getInternalState(moduleProvider, "notifyHandler");
+ final HttpClient client = ReflectUtil.getInternalState(handler, "httpClient");
+ assertEquals(Duration.ofSeconds(3), client.connectTimeout().orElseThrow());
+ final List<HttpAlarmCallback> hooks = handler.buildCallbacks()
+ .stream()
+ .filter(HttpAlarmCallback.class::isInstance)
+ .map(HttpAlarmCallback.class::cast)
+ .collect(Collectors.toList());
+ assertEquals(8, hooks.size());
+ for (final HttpAlarmCallback hook : hooks) {
+ final String name = hook.getClass().getSimpleName();
+ assertSame(client, ReflectUtil.getInternalState(hook, "httpClient"), name);
+ assertEquals(Duration.ofSeconds(3), ReflectUtil.getInternalState(hook, "timeout"), name);
+ }
+ }
}
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandlerTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandlerTest.java
index 3c80dbd7c5..3d7cc69fa4 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandlerTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandlerTest.java
@@ -45,6 +45,8 @@ import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import org.apache.skywalking.oap.server.testing.util.ReflectUtil;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -262,7 +264,8 @@ public class NotifyHandlerTest {
ModuleManager moduleManager = mock(ModuleManager.class);
- notifyHandler = new NotifyHandler(new AlarmRulesWatcher(rules, null, moduleManager), moduleManager);
+ notifyHandler = new NotifyHandler(
+ new AlarmRulesWatcher(rules, null, moduleManager), moduleManager, HttpClient.newHttpClient(), Duration.ofSeconds(12));
notifyHandler.init(new AlarmCallback() {
@Override
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallbackTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallbackTest.java
index dd0ab94094..8e7fb61b28 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallbackTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/dingtalk/DingtalkHookCallbackTest.java
@@ -34,6 +34,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -91,7 +93,7 @@ public class DingtalkHookCallbackTest {
rules.getDingtalkSettingsMap().put(setting2.getFormattedName(), setting2);
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- DingtalkHookCallback dingtalkCallBack = new DingtalkHookCallback(alarmRulesWatcher);
+ DingtalkHookCallback dingtalkCallBack = new DingtalkHookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
alarmMessage.setScopeId(DefaultScopeDefine.SERVICE);
@@ -126,7 +128,7 @@ public class DingtalkHookCallbackTest {
rules.getDingtalkSettingsMap().put(setting2.getFormattedName(), setting2);
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- DingtalkHookCallback dingtalkCallBack = new DingtalkHookCallback(alarmRulesWatcher);
+ DingtalkHookCallback dingtalkCallBack = new DingtalkHookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
alarmMessage.setScopeId(DefaultScopeDefine.SERVICE);
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/feishu/FeishuHookCallbackTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/feishu/FeishuHookCallbackTest.java
index 0accf98d1c..e9dc4845e9 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/feishu/FeishuHookCallbackTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/feishu/FeishuHookCallbackTest.java
@@ -34,6 +34,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -89,7 +91,7 @@ public class FeishuHookCallbackTest {
rules.getFeishuSettingsMap().put(setting2.getFormattedName(), setting2);
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- FeishuHookCallback feishuHookCallback = new FeishuHookCallback(alarmRulesWatcher);
+ FeishuHookCallback feishuHookCallback = new FeishuHookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
alarmMessage.setScopeId(DefaultScopeDefine.SERVICE);
@@ -124,7 +126,7 @@ public class FeishuHookCallbackTest {
rules.getFeishuSettingsMap().put(setting2.getFormattedName(), setting2);
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- FeishuHookCallback feishuHookCallback = new FeishuHookCallback(alarmRulesWatcher);
+ FeishuHookCallback feishuHookCallback = new FeishuHookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
alarmMessage.setScopeId(DefaultScopeDefine.SERVICE);
@@ -159,7 +161,7 @@ public class FeishuHookCallbackTest {
rules.getFeishuSettingsMap().put(setting2.getFormattedName(), setting2);
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- FeishuHookCallback feishuHookCallback = new FeishuHookCallback(alarmRulesWatcher);
+ FeishuHookCallback feishuHookCallback = new FeishuHookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
alarmMessage.setScopeId(DefaultScopeDefine.SERVICE);
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/pagerduty/PagerDutyHookCallbackTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/pagerduty/PagerDutyHookCallbackTest.java
index 9442e292af..1c2ae91247 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/pagerduty/PagerDutyHookCallbackTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/pagerduty/PagerDutyHookCallbackTest.java
@@ -18,6 +18,8 @@
package org.apache.skywalking.oap.server.core.alarm.provider.pagerduty;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.ArrayList;
import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
import org.apache.skywalking.oap.server.core.alarm.provider.AlarmHooksType;
@@ -90,7 +92,7 @@ public class PagerDutyHookCallbackTest {
alarmMessage.setAlarmMessage("alarmMessage with [DefaultScopeDefine.All]");
alarmMessage.getHooks().add(setting.getFormattedName());
- new PagerDutyHookCallback(new AlarmRulesWatcher(rules, null, null))
+ new PagerDutyHookCallback(new AlarmRulesWatcher(rules, null, null), HttpClient.newHttpClient(), Duration.ofSeconds(12))
.doAlarm(new ArrayList<>(List.of(alarmMessage)));
// one request per integration key
@@ -124,7 +126,7 @@ public class PagerDutyHookCallbackTest {
rules.getPagerDutySettingsMap().put(setting1.getFormattedName(), setting1);
rules.getPagerDutySettingsMap().put(setting2.getFormattedName(), setting2);
PagerDutyHookCallback pagerDutyHookCallback = new PagerDutyHookCallback(
- new AlarmRulesWatcher(rules, null, null)
+ new AlarmRulesWatcher(rules, null, null), HttpClient.newHttpClient(), Duration.ofSeconds(12)
);
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/slack/SlackHookCallbackTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/slack/SlackHookCallbackTest.java
index 5cac6ec355..edd848caea 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/slack/SlackHookCallbackTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/slack/SlackHookCallbackTest.java
@@ -24,6 +24,8 @@ import com.linecorp.armeria.common.HttpResponse;
import com.linecorp.armeria.common.HttpStatus;
import com.linecorp.armeria.server.ServerBuilder;
import com.linecorp.armeria.testing.junit5.server.ServerExtension;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -79,7 +81,7 @@ public class SlackHookCallbackTest {
rules.getSlackSettingsMap().put(setting1.getFormattedName(), setting1);
rules.getSlackSettingsMap().put(setting2.getFormattedName(), setting2);
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- SlackhookCallback slackhookCallback = new SlackhookCallback(alarmRulesWatcher);
+ SlackhookCallback slackhookCallback = new SlackhookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
alarmMessage.setScopeId(DefaultScopeDefine.SERVICE);
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/webhook/WebhookCallbackTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/webhook/WebhookCallbackTest.java
index b9770ef3c9..70ad653611 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/webhook/WebhookCallbackTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/webhook/WebhookCallbackTest.java
@@ -36,6 +36,8 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
@@ -112,7 +114,7 @@ public class WebhookCallbackTest {
rules.getWebhookSettingsMap().put(setting2.getFormattedName(), setting2);
setting2.setHeaders(ImmutableMap.of("Authorization", " Bearer bearer_token", "x-company-header", "arbitrary-additional-http-headers"));
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- WebhookCallback webhookCallback = new WebhookCallback(alarmRulesWatcher);
+ WebhookCallback webhookCallback = new WebhookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
alarmMessage.setId0("1");
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/wechat/WechatHookCallbackTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/wechat/WechatHookCallbackTest.java
index c70b776365..4470c0c5bd 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/wechat/WechatHookCallbackTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/wechat/WechatHookCallbackTest.java
@@ -34,6 +34,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -90,7 +92,7 @@ public class WechatHookCallbackTest {
rules.getWechatSettingsMap().put(setting1.getFormattedName(), setting1);
rules.getWechatSettingsMap().put(setting2.getFormattedName(), setting2);
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- WechatHookCallback wechatHookCallback = new WechatHookCallback(alarmRulesWatcher);
+ WechatHookCallback wechatHookCallback = new WechatHookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
List<AlarmMessage> alarmRecoveryMessages = new ArrayList<>(1);
AlarmMessage alarmMessage = new AlarmMessage();
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/welink/WeLinkHookCallbackTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/welink/WeLinkHookCallbackTest.java
index 511f0a3cfa..43528fa855 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/welink/WeLinkHookCallbackTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/welink/WeLinkHookCallbackTest.java
@@ -35,6 +35,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -90,7 +92,7 @@ public class WeLinkHookCallbackTest {
rules.getWeLinkSettingsMap().put(setting2.getFormattedName(), setting2);
AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(rules, null, null);
- WeLinkHookCallback welinkHookCallback = new WeLinkHookCallback(alarmRulesWatcher);
+ WeLinkHookCallback welinkHookCallback = new WeLinkHookCallback(alarmRulesWatcher, HttpClient.newHttpClient(), Duration.ofSeconds(12));
List<AlarmMessage> alarmMessages = new ArrayList<>(2);
AlarmMessage alarmMessage = new AlarmMessage();
alarmMessage.setScopeId(DefaultScopeDefine.SERVICE);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/HttpAlarmCallback.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/HttpAlarmCallback.java
index c5d4965baf..39c52224e7 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/HttpAlarmCallback.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/HttpAlarmCallback.java
@@ -18,22 +18,39 @@
package org.apache.skywalking.oap.server.core.alarm;
+import org.apache.skywalking.oap.server.library.util.JdkHttpClientUtils;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.List;
import java.util.Map;
public abstract class HttpAlarmCallback implements AlarmCallback {
- private static final HttpClient HTTP_CLIENT = HttpClient
+ private final HttpClient httpClient;
+
+ private final Duration timeout;
+
+ /**
+ * @param httpClient shared by every HTTP hook, see {@link #newHttpClient(Duration)}.
+ * @param timeout deadline of each {@link #post(URI, String, Map)}, covering connect, the response headers and
+ * the response body.
+ */
+ protected HttpAlarmCallback(final HttpClient httpClient, final Duration timeout) {
+ this.httpClient = httpClient;
+ this.timeout = timeout;
+ }
+
+ public static HttpClient newHttpClient(final Duration connectTimeout) {
+ return HttpClient
.newBuilder()
.followRedirects(HttpClient.Redirect.NORMAL)
+ .connectTimeout(connectTimeout)
.build();
+ }
protected String post(
final URI uri,
@@ -45,11 +62,12 @@ public abstract class HttpAlarmCallback implements AlarmCallback {
.uri(uri)
.POST(HttpRequest.BodyPublishers.ofString(body))
.header("Content-Type", "application/json")
- .timeout(Duration.ofSeconds(12));
+ // JdkHttpClientUtils requires it: only the request's own timeout aborts an exchange still waiting
+ // for the response headers.
+ .timeout(timeout);
headers.forEach(request::header);
- final var response = HTTP_CLIENT
- .send(request.build(), HttpResponse.BodyHandlers.ofString());
+ final var response = JdkHttpClientUtils.sendStringWithTimeout(httpClient, request.build(), timeout);
final var status = response.statusCode();
// Any 2xx means the hook endpoint accepted the alarm. Asynchronous intake APIs answer with 202 rather
@@ -79,5 +97,4 @@ public abstract class HttpAlarmCallback implements AlarmCallback {
}
protected abstract void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) throws Exception ;
-
}
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/alarm/HttpAlarmCallbackTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/alarm/HttpAlarmCallbackTest.java
index 42c724c356..9b0f1c8e7d 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/alarm/HttpAlarmCallbackTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/alarm/HttpAlarmCallbackTest.java
@@ -33,11 +33,19 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
+import java.net.http.HttpTimeoutException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class HttpAlarmCallbackTest {
@@ -63,6 +71,14 @@ public class HttpAlarmCallbackTest {
}
private static final class Callback extends HttpAlarmCallback {
+ Callback() {
+ this(Duration.ofSeconds(12));
+ }
+
+ Callback(final Duration timeout) {
+ super(HttpAlarmCallback.newHttpClient(timeout), timeout);
+ }
+
@Override
protected void doAlarmCallback(List<AlarmMessage> alarmMessages, boolean isRecovery) {
}
@@ -73,6 +89,10 @@ public class HttpAlarmCallbackTest {
"{}", Map.of()
);
}
+
+ String send(final URI uri) throws IOException, InterruptedException {
+ return post(uri, "{}", Map.of());
+ }
}
@BeforeAll
@@ -132,4 +152,36 @@ public class HttpAlarmCallbackTest {
assertEquals(Level.ERROR, APPENDER.events.get(0).getLevel());
assertTrue(APPENDER.events.get(0).getMessage().getFormattedMessage().contains("500"));
}
-}
\ No newline at end of file
+
+ @Test
+ public void stalledResponseBodyTimesOut() throws Exception {
+ // The request's own timeout stops at the response headers, so only the delivery deadline bounds the body.
+ final CountDownLatch release = new CountDownLatch(1);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final HttpServer stalling = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ stalling.setExecutor(executor);
+ stalling.createContext("/stall", exchange -> {
+ exchange.sendResponseHeaders(200, 64);
+ exchange.getResponseBody().write("ping".getBytes(StandardCharsets.UTF_8));
+ exchange.getResponseBody().flush();
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ exchange.close();
+ });
+ stalling.start();
+ try {
+ final URI uri = URI.create("http://127.0.0.1:" + stalling.getAddress().getPort() + "/stall");
+ final long started = System.nanoTime();
+ assertThrows(HttpTimeoutException.class, () -> new Callback(Duration.ofSeconds(1)).send(uri));
+ final long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
+ assertTrue(elapsedMs < 5_000, "elapsedMs=" + elapsedMs);
+ } finally {
+ release.countDown();
+ stalling.stop(0);
+ executor.shutdownNow();
+ }
+ }
+}
diff --git a/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/JdkHttpClientUtils.java b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/JdkHttpClientUtils.java
new file mode 100644
index 0000000000..42456407b2
--- /dev/null
+++ b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/JdkHttpClientUtils.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.skywalking.oap.server.library.util;
+
+import java.io.IOException;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.net.http.HttpTimeoutException;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Flow;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * JDK {@link HttpClient} helpers.
+ * Configuration defaults, status handling, parsing, and retries stay with the caller.
+ */
+public final class JdkHttpClientUtils {
+ private JdkHttpClientUtils() {
+ }
+
+ /**
+ * Send {@code request} and read the whole response body as a String.
+ * {@code timeout} is one budget for the entire operation: time spent waiting for the
+ * response headers is subtracted from the time left to read the body.
+ * {@link HttpRequest.Builder#timeout(Duration)} does not cover body consumption.
+ *
+ * <p>{@code request} must carry its own {@link HttpRequest.Builder#timeout(Duration)}, normally the
+ * same {@code timeout}: on Java 11 cancelling the pending exchange releases the caller but does not
+ * abort a request still waiting for the response headers, only the request's own timeout does.
+ */
+ public static HttpResponse<String> sendStringWithTimeout(final HttpClient client,
+ final HttpRequest request,
+ final Duration timeout) throws IOException, InterruptedException {
+ if (timeout.isNegative() || timeout.isZero()) {
+ throw new IllegalArgumentException("timeout must be positive");
+ }
+ if (request.timeout().isEmpty()) {
+ throw new IllegalArgumentException("request must carry its own timeout");
+ }
+ final long budgetNanos = budgetNanos(timeout);
+
+ final CancellableBody handler = new CancellableBody();
+ final long started = System.nanoTime();
+ final CompletableFuture<HttpResponse<String>> pending = client.sendAsync(request, handler);
+
+ try {
+ final long remaining = budgetNanos - (System.nanoTime() - started);
+ if (remaining <= 0) {
+ throw new TimeoutException();
+ }
+ return pending.get(remaining, TimeUnit.NANOSECONDS);
+ } catch (TimeoutException e) {
+ final HttpTimeoutException failure = new HttpTimeoutException("HTTP request timed out");
+ failure.initCause(e);
+ handler.cancel(failure);
+ pending.cancel(true);
+ throw failure;
+ } catch (InterruptedException e) {
+ handler.cancel(e);
+ pending.cancel(true);
+ throw e;
+ } catch (ExecutionException e) {
+ final Throwable cause = e.getCause();
+ handler.cancel(cause);
+ pending.cancel(true);
+ if (cause instanceof IOException) {
+ throw (IOException) cause;
+ }
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new IOException("HTTP request failed", cause);
+ }
+ }
+
+ private static long budgetNanos(final Duration positiveTimeout) {
+ try {
+ return positiveTimeout.toNanos();
+ } catch (ArithmeticException overflow) {
+ return Long.MAX_VALUE;
+ }
+ }
+
+ /**
+ * One instance per request. Cancelling the subscriber closes a stalled body read.
+ * On Java 11, cancelling only the request future releases the caller but leaves the connection open.
+ */
+ private static final class CancellableBody implements HttpResponse.BodyHandler<String>, HttpResponse.BodySubscriber<String> {
+ private static final Flow.Subscription CANCELLED = new Flow.Subscription() {
+ @Override
+ public void request(final long n) {
+ }
+
+ @Override
+ public void cancel() {
+ }
+ };
+
+ private final AtomicReference<Flow.Subscription> subscription = new AtomicReference<>();
+
+ private final CompletableFuture<String> body = new CompletableFuture<>();
+
+ private HttpResponse.BodySubscriber<String> delegate;
+
+ @Override
+ public HttpResponse.BodySubscriber<String> apply(final HttpResponse.ResponseInfo info) {
+ // Preserve the existing response charset handling.
+ delegate = HttpResponse.BodyHandlers.ofString().apply(info);
+ delegate.getBody().whenComplete((value, error) -> {
+ if (error == null) {
+ body.complete(value);
+ } else {
+ body.completeExceptionally(error);
+ }
+ });
+ return this;
+ }
+
+ @Override
+ public CompletableFuture<String> getBody() {
+ return body;
+ }
+
+ @Override
+ public void onSubscribe(final Flow.Subscription incoming) {
+ if (subscription.compareAndSet(null, incoming)) {
+ delegate.onSubscribe(incoming);
+ } else {
+ // Includes cancellation before the subscription arrives.
+ incoming.cancel();
+ }
+ }
+
+ @Override
+ public void onNext(final List<ByteBuffer> data) {
+ delegate.onNext(data);
+ }
+
+ @Override
+ public void onError(final Throwable error) {
+ delegate.onError(error);
+ }
+
+ @Override
+ public void onComplete() {
+ delegate.onComplete();
+ }
+
+ void cancel(final Throwable failure) {
+ final Flow.Subscription previous = subscription.getAndSet(CANCELLED);
+ if (previous != null && previous != CANCELLED) {
+ previous.cancel();
+ }
+ // Cancellation need not trigger onError/onComplete.
+ body.completeExceptionally(failure);
+ }
+ }
+}
diff --git a/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/JdkHttpClientUtilsTest.java b/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/JdkHttpClientUtilsTest.java
new file mode 100644
index 0000000000..d5c0274bdf
--- /dev/null
+++ b/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/JdkHttpClientUtilsTest.java
@@ -0,0 +1,426 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.skywalking.oap.server.library.util;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
+import java.net.Authenticator;
+import java.net.CookieHandler;
+import java.net.InetAddress;
+import java.net.ProxySelector;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.net.http.HttpTimeoutException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Flow;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLParameters;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Timeout(value = 20, unit = TimeUnit.SECONDS)
+public class JdkHttpClientUtilsTest {
+
+ @Test
+ public void rejectsNonPositiveTimeout() {
+ HttpClient client = HttpClient.newHttpClient();
+ HttpRequest request = HttpRequest.newBuilder(URI.create("http://127.0.0.1/")).timeout(Duration.ofSeconds(1)).build();
+ assertThrows(IllegalArgumentException.class, () -> JdkHttpClientUtils.sendStringWithTimeout(client, request, Duration.ZERO));
+ assertThrows(IllegalArgumentException.class, () -> JdkHttpClientUtils.sendStringWithTimeout(client, request, Duration.ofMillis(-1)));
+ // Overflows Duration.toNanos(); must not saturate into an effectively unlimited timeout.
+ assertThrows(IllegalArgumentException.class, () -> JdkHttpClientUtils.sendStringWithTimeout(client, request, Duration.ofSeconds(-10_000_000_000L)));
+ }
+
+ @Test
+ public void rejectsRequestWithoutItsOwnTimeout() {
+ HttpClient client = HttpClient.newHttpClient();
+ HttpRequest request = HttpRequest.newBuilder(URI.create("http://127.0.0.1/")).build();
+ assertThrows(IllegalArgumentException.class, () -> JdkHttpClientUtils.sendStringWithTimeout(client, request, Duration.ofSeconds(1)));
+ }
+
+ @Test
+ public void extremeTimeoutDoesNotOverflowToNanos() throws Exception {
+ try (RawHttp peer = new RawHttp(RawHttp.Mode.CHARSET, 0)) {
+ // Larger than Long.MAX_VALUE nanoseconds; the util must not throw ArithmeticException.
+ HttpResponse<String> response = send(peer.uri(), Duration.ofSeconds(10_000_000_000L));
+ assertEquals("\u00e9", response.body());
+ }
+ }
+
+ @Test
+ public void delayedHeadersTimeOutAndCloseTheConnection() throws Exception {
+ try (RawHttp peer = new RawHttp(RawHttp.Mode.DELAY_HEADERS, 5_000)) {
+ long started = System.nanoTime();
+ // Do not assert the exception message: HttpRequest.timeout and the util share the same
+ // deadline, so either "HTTP request timed out" or the JDK's "request timed out" may win.
+ assertThrows(HttpTimeoutException.class, () -> send(peer.uri(), Duration.ofSeconds(1)));
+ long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
+ assertTrue(elapsedMs >= 700 && elapsedMs < 2_500, "elapsedMs=" + elapsedMs);
+ assertTrue(peer.awaitClosed(3, TimeUnit.SECONDS), "stalled connection was not closed");
+ }
+ }
+
+ @Test
+ public void stalledBodyTimesOutAndClosesTheConnection() throws Exception {
+ try (RawHttp peer = new RawHttp(RawHttp.Mode.STALL_BODY, 0)) {
+ long started = System.nanoTime();
+ assertThrows(HttpTimeoutException.class, () -> send(peer.uri(), Duration.ofSeconds(1)));
+ long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
+ assertTrue(elapsedMs >= 700 && elapsedMs < 2_500, "elapsedMs=" + elapsedMs);
+ assertTrue(peer.awaitClosed(3, TimeUnit.SECONDS), "stalled body connection was not closed");
+ }
+ }
+
+ @Test
+ public void headersAndBodyShareOneDeadline() throws Exception {
+ try (RawHttp peer = new RawHttp(RawHttp.Mode.DELAY_THEN_STALL, 1_200)) {
+ long started = System.nanoTime();
+ assertThrows(HttpTimeoutException.class, () -> send(peer.uri(), Duration.ofSeconds(2)));
+ long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started);
+ // A fresh 2s body budget after the 1.2s header delay would finish near 3200ms.
+ assertTrue(elapsedMs >= 1_600 && elapsedMs < 2_900, "elapsedMs=" + elapsedMs);
+ assertTrue(peer.awaitClosed(3, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ public void interruptionCancelsTheBodyAndClosesTheConnection() throws Exception {
+ try (RawHttp peer = new RawHttp(RawHttp.Mode.STALL_BODY, 0)) {
+ SubscriptionSignallingClient client = new SubscriptionSignallingClient(newClient());
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ Thread caller = new Thread(() -> {
+ try {
+ send(client, peer.uri(), Duration.ofSeconds(30));
+ } catch (Throwable thrown) {
+ failure.set(thrown);
+ }
+ });
+ caller.start();
+ // The body subscriber holds its subscription, so the interruption hits the body read.
+ assertTrue(client.subscribed.await(5, TimeUnit.SECONDS));
+ caller.interrupt();
+ caller.join(5_000);
+ assertInstanceOf(InterruptedException.class, failure.get());
+ assertTrue(peer.awaitClosed(3, TimeUnit.SECONDS), "interrupted request left the connection open");
+ }
+ }
+
+ @Test
+ public void cancelBeforeSubscriptionArrivesCancelsTheSubscription() throws Exception {
+ Class<?> type = Class.forName(JdkHttpClientUtils.class.getName() + "$CancellableBody");
+ Constructor<?> constructor = type.getDeclaredConstructor();
+ constructor.setAccessible(true);
+ Object handler = constructor.newInstance();
+ Method cancel = type.getDeclaredMethod("cancel", Throwable.class);
+ cancel.setAccessible(true);
+ cancel.invoke(handler, new HttpTimeoutException("HTTP request timed out"));
+
+ AtomicBoolean cancelled = new AtomicBoolean();
+ Flow.Subscription incoming = new Flow.Subscription() {
+ @Override
+ public void request(long n) {
+ }
+
+ @Override
+ public void cancel() {
+ cancelled.set(true);
+ }
+ };
+ Method onSubscribe = type.getDeclaredMethod("onSubscribe", Flow.Subscription.class);
+ onSubscribe.setAccessible(true);
+ onSubscribe.invoke(handler, incoming);
+ assertTrue(cancelled.get());
+ }
+
+ @Test
+ public void preservesTheResponseCharset() throws Exception {
+ try (RawHttp peer = new RawHttp(RawHttp.Mode.CHARSET, 0)) {
+ HttpResponse<String> response = send(peer.uri(), Duration.ofSeconds(2));
+ assertEquals("\u00e9", response.body());
+ }
+ }
+
+ private static HttpResponse<String> send(URI uri, Duration timeout) throws IOException, InterruptedException {
+ return send(newClient(), uri, timeout);
+ }
+
+ private static HttpResponse<String> send(HttpClient client, URI uri, Duration timeout) throws IOException, InterruptedException {
+ HttpRequest request = HttpRequest.newBuilder(uri).timeout(timeout).GET().build();
+ return JdkHttpClientUtils.sendStringWithTimeout(client, request, timeout);
+ }
+
+ private static HttpClient newClient() {
+ return HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).connectTimeout(Duration.ofSeconds(2)).build();
+ }
+
+ /**
+ * A real client that opens {@code subscribed} once the body subscriber passed by
+ * {@link JdkHttpClientUtils} has received its subscription, that is, once the body is being read.
+ */
+ private static final class SubscriptionSignallingClient extends HttpClient {
+ private final HttpClient delegate;
+
+ private final CountDownLatch subscribed = new CountDownLatch(1);
+
+ private SubscriptionSignallingClient(HttpClient delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest request, HttpResponse.BodyHandler<T> handler) {
+ return delegate.sendAsync(request, info -> {
+ HttpResponse.BodySubscriber<T> body = handler.apply(info);
+ return new HttpResponse.BodySubscriber<T>() {
+ @Override
+ public CompletionStage<T> getBody() {
+ return body.getBody();
+ }
+
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ body.onSubscribe(subscription);
+ subscribed.countDown();
+ }
+
+ @Override
+ public void onNext(List<ByteBuffer> item) {
+ body.onNext(item);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ body.onError(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ body.onComplete();
+ }
+ };
+ });
+ }
+
+ @Override
+ public <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest request, HttpResponse.BodyHandler<T> handler,
+ HttpResponse.PushPromiseHandler<T> pushPromiseHandler) {
+ return delegate.sendAsync(request, handler, pushPromiseHandler);
+ }
+
+ @Override
+ public <T> HttpResponse<T> send(HttpRequest request, HttpResponse.BodyHandler<T> handler) throws IOException, InterruptedException {
+ return delegate.send(request, handler);
+ }
+
+ @Override
+ public Optional<CookieHandler> cookieHandler() {
+ return delegate.cookieHandler();
+ }
+
+ @Override
+ public Optional<Duration> connectTimeout() {
+ return delegate.connectTimeout();
+ }
+
+ @Override
+ public Redirect followRedirects() {
+ return delegate.followRedirects();
+ }
+
+ @Override
+ public Optional<ProxySelector> proxy() {
+ return delegate.proxy();
+ }
+
+ @Override
+ public SSLContext sslContext() {
+ return delegate.sslContext();
+ }
+
+ @Override
+ public SSLParameters sslParameters() {
+ return delegate.sslParameters();
+ }
+
+ @Override
+ public Optional<Authenticator> authenticator() {
+ return delegate.authenticator();
+ }
+
+ @Override
+ public Version version() {
+ return delegate.version();
+ }
+
+ @Override
+ public Optional<Executor> executor() {
+ return delegate.executor();
+ }
+ }
+
+ /**
+ * Minimal HTTP/1.1 peer. A stalled response keeps the accepted socket open until the client cancels it.
+ */
+ private static final class RawHttp implements AutoCloseable {
+ private enum Mode {
+ DELAY_HEADERS, STALL_BODY, DELAY_THEN_STALL, CHARSET
+ }
+
+ private final ServerSocket server;
+
+ private final Thread thread;
+
+ private final CountDownLatch closed = new CountDownLatch(1);
+
+ private volatile Socket socket;
+
+ private RawHttp(Mode mode, long delayMs) throws IOException {
+ server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"));
+ thread = new Thread(() -> serve(mode, delayMs), "jdk-http-utils-test");
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ private URI uri() {
+ return URI.create("http://127.0.0.1:" + server.getLocalPort() + "/");
+ }
+
+ private boolean awaitClosed(long time, TimeUnit unit) throws InterruptedException {
+ return closed.await(time, unit);
+ }
+
+ private void serve(Mode mode, long delayMs) {
+ try {
+ socket = server.accept();
+ socket.setTcpNoDelay(true);
+ InputStream in = socket.getInputStream();
+ OutputStream out = socket.getOutputStream();
+ if (!drainRequest(in)) {
+ return;
+ }
+ if (mode == Mode.DELAY_HEADERS) {
+ if (sleepOrClosed(in, delayMs)) {
+ return;
+ }
+ out.write(ascii("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok"));
+ out.flush();
+ return;
+ }
+ if (mode == Mode.DELAY_THEN_STALL && sleepOrClosed(in, delayMs)) {
+ return;
+ }
+ if (mode == Mode.CHARSET) {
+ out.write(ascii("HTTP/1.1 200 OK\r\nContent-Type: text/plain; charset=ISO-8859-1\r\nContent-Length: 1\r\nConnection: close\r\n\r\n"));
+ out.write(0xE9);
+ out.flush();
+ waitForClose(in);
+ return;
+ }
+ out.write(ascii("HTTP/1.1 200 OK\r\nContent-Type: text/plain; charset=UTF-8\r\nContent-Length: 64\r\nConnection: keep-alive\r\n\r\n"));
+ out.write(ascii("ping"));
+ out.flush();
+ waitForClose(in);
+ } catch (IOException ignored) {
+ closed.countDown();
+ }
+ }
+
+ private boolean sleepOrClosed(InputStream in, long delayMs) throws IOException {
+ long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(delayMs);
+ socket.setSoTimeout(200);
+ while (System.nanoTime() < deadline) {
+ try {
+ if (in.read() < 0) {
+ closed.countDown();
+ return true;
+ }
+ } catch (SocketTimeoutException ignored) {
+ // Peer is still open.
+ }
+ }
+ socket.setSoTimeout(0);
+ return false;
+ }
+
+ private void waitForClose(InputStream in) throws IOException {
+ socket.setSoTimeout(0);
+ try {
+ if (in.read() < 0) {
+ closed.countDown();
+ }
+ } catch (IOException e) {
+ closed.countDown();
+ }
+ }
+
+ private boolean drainRequest(InputStream in) throws IOException {
+ byte[] end = new byte[] {
+ '\r', '\n', '\r', '\n'
+ };
+ int matched = 0;
+ int value;
+ while ((value = in.read()) >= 0) {
+ matched = value == end[matched] ? matched + 1 : 0;
+ if (matched == end.length) {
+ return true;
+ }
+ }
+ closed.countDown();
+ return false;
+ }
+
+ private static byte[] ascii(String text) {
+ return text.getBytes(StandardCharsets.US_ASCII);
+ }
+
+ @Override
+ public void close() throws IOException {
+ server.close();
+ Socket current = socket;
+ if (current != null) {
+ current.close();
+ }
+ thread.interrupt();
+ }
+ }
+}
diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml
index 92084a815f..683703a9de 100644
--- a/oap-server/server-starter/src/main/resources/application.yml
+++ b/oap-server/server-starter/src/main/resources/application.yml
@@ -599,6 +599,9 @@ traceQL:
alarm:
selector: ${SW_ALARM:default}
default:
+ # Deadline (seconds) of each HTTP request an HTTP-based alarm hook (every hook except gRPC) sends,
+ # covering connecting, waiting for the response and reading the response body. Values <= 0 fall back to 12.
+ httpTimeout: ${SW_ALARM_HTTP_TIMEOUT:12}
telemetry:
selector: ${SW_TELEMETRY:prometheus}
diff --git a/oap-server/server-tools/data-generator/src/main/resources/application.yml b/oap-server/server-tools/data-generator/src/main/resources/application.yml
index d20b5971a3..66774c470a 100755
--- a/oap-server/server-tools/data-generator/src/main/resources/application.yml
+++ b/oap-server/server-tools/data-generator/src/main/resources/application.yml
@@ -222,6 +222,9 @@ query:
alarm:
selector: ${SW_ALARM:default}
default:
+ # Deadline (seconds) of each HTTP request an HTTP-based alarm hook (every hook except gRPC) sends,
+ # covering connecting, waiting for the response and reading the response body. Values <= 0 fall back to 12.
+ httpTimeout: ${SW_ALARM_HTTP_TIMEOUT:12}
telemetry:
selector: ${SW_TELEMETRY:none}
diff --git a/test/e2e-v2/cases/storage/expected/config-dump.yml b/test/e2e-v2/cases/storage/expected/config-dump.yml
index b2f4c07d83..7022306be5 100644
--- a/test/e2e-v2/cases/storage/expected/config-dump.yml
+++ b/test/e2e-v2/cases/storage/expected/config-dump.yml
@@ -55,6 +55,7 @@
"ai-pipeline.default.uriRecognitionServerAddr": "",
"ai-pipeline.default.uriRecognitionServerPort": "17128",
"ai-pipeline.provider": "default",
+ "alarm.default.httpTimeout": "12",
"alarm.provider": "default",
"aws-firehose.default.acceptQueueSize": "0",
"aws-firehose.default.contextPath": "/",