From e9a0598261fdf715e9c940e413abf01f743a6220 Mon Sep 17 00:00:00 2001 From: Duansg Date: Wed, 29 Jul 2026 03:51:02 -0700 Subject: [PATCH 1/2] [fix] require an account to subscribe to the alert and manager streams --- .../alert/config/AlertSseManager.java | 31 +++- .../alert/config/AlertSseManagerTest.java | 56 +++++++ .../manager/config/ManagerSseManager.java | 31 +++- .../manager/config/ManagerSseManagerTest.java | 65 ++++++++ .../src/main/resources/sureness.yml | 6 +- .../startup/security/SurenessSseRuleTest.java | 98 ++++++++++++ .../hertzbeat-mysql-iotdb/conf/sureness.yml | 6 +- .../conf/sureness.yml | 6 +- .../conf/sureness.yml | 6 +- .../conf/sureness.yml | 6 +- .../conf/sureness.yml | 6 +- script/sureness.yml | 6 +- .../layout/basic/widgets/notify.component.ts | 142 +++++++++--------- .../alert-center/alert-center.component.ts | 38 ++--- .../service/authorized-sse.service.spec.ts | 103 +++++++++++++ .../src/app/service/authorized-sse.service.ts | 109 ++++++++++++++ 16 files changed, 610 insertions(+), 105 deletions(-) create mode 100644 hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java create mode 100644 hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java create mode 100644 web-app/src/app/service/authorized-sse.service.spec.ts create mode 100644 web-app/src/app/service/authorized-sse.service.ts diff --git a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java index e3d245621e5..461846b5800 100644 --- a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java +++ b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java @@ -19,9 +19,12 @@ package org.apache.hertzbeat.alert.config; +import lombok.Setter; import lombok.extern.slf4j.Slf4j; +import org.springframework.http.HttpStatus; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; +import org.springframework.web.server.ResponseStatusException; import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @@ -36,10 +39,32 @@ @Slf4j @Component public class AlertSseManager { + + /** + * How long a subscription may stay open before the client has to reconnect. + * + *

`Long.MAX_VALUE` meant a subscription never expired on its own, so a client that + * went away without closing cleanly held its request thread until the container noticed. + * A finite timeout bounds that; browsers reconnect on timeout, and the ui re-subscribes. + */ + private static final long EMITTER_TIMEOUT_MILLIS = 30 * 60 * 1000L; + + /** + * Cap on concurrently held subscriptions. Each one occupies a request thread, so without + * a ceiling enough parallel subscriptions exhaust the container's thread pool and take + * the whole application down with them. + */ + @Setter + private int maxEmitters = 1000; + private final Map emitters = new ConcurrentHashMap<>(); public SseEmitter createEmitter(Long clientId) { - SseEmitter emitter = new SseEmitter(Long.MAX_VALUE); + if (emitters.size() >= maxEmitters) { + log.warn("Refused alert subscription, already holding {} of at most {}", emitters.size(), maxEmitters); + throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "Too many alert subscriptions"); + } + SseEmitter emitter = new SseEmitter(EMITTER_TIMEOUT_MILLIS); emitter.onCompletion(() -> removeEmitter(clientId)); emitter.onTimeout(() -> removeEmitter(clientId)); emitter.onError((ex) -> removeEmitter(clientId)); @@ -47,6 +72,10 @@ public SseEmitter createEmitter(Long clientId) { return emitter; } + int subscriptionCount() { + return emitters.size(); + } + @Async public void broadcast(String data) { emitters.forEach((clientId, emitter) -> { diff --git a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java index 40e88ddef99..3fd3cdf23f6 100644 --- a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java +++ b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java @@ -19,15 +19,19 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.http.HttpStatus; +import org.springframework.web.server.ResponseStatusException; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.lang.reflect.Field; import java.util.HashMap; import java.util.Map; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -66,4 +70,56 @@ void testCompleteThrowsException() throws Exception { assertFalse(currentEmitters.containsKey(1L), "Emitter should still exist because complete() threw exception"); } + /** + * An unbounded emitter never expires on its own, so a client that goes away without + * closing cleanly keeps holding its request thread until the container notices. + */ + @Test + void testSubscriptionsAreGivenFiniteTimeout() { + SseEmitter emitter = alertSseManager.createEmitter(1L); + + assertNotNull(emitter.getTimeout()); + assertTrue(emitter.getTimeout() > 0 && emitter.getTimeout() < Long.MAX_VALUE, + "timeout must be finite, was " + emitter.getTimeout()); + } + + /** + * Each open subscription occupies a request thread, so enough of them in parallel + * exhaust the container's pool and take the whole application down. + */ + @Test + void testSubscriptionsBeyondLimitAreRefused() { + alertSseManager.setMaxEmitters(2); + + alertSseManager.createEmitter(1L); + alertSseManager.createEmitter(2L); + ResponseStatusException thrown = + assertThrows(ResponseStatusException.class, () -> alertSseManager.createEmitter(3L)); + + assertEquals(HttpStatus.SERVICE_UNAVAILABLE, thrown.getStatusCode()); + assertEquals(2, alertSseManager.subscriptionCount()); + } + + /** + * The cap must not become a permanent lockout: once a dead subscription is cleaned up, + * its slot has to be available again. + */ + @Test + void testDroppedSubscriptionFreesItsSlot() throws Exception { + alertSseManager.setMaxEmitters(1); + alertSseManager.createEmitter(1L); + assertThrows(ResponseStatusException.class, () -> alertSseManager.createEmitter(2L)); + + // a client that went away makes the next send fail, which is how the manager notices + SseEmitter deadEmitter = mock(SseEmitter.class); + doThrow(new IllegalStateException("client gone")).when(deadEmitter).send(any(SseEmitter.SseEventBuilder.class)); + Field emittersField = AlertSseManager.class.getDeclaredField("emitters"); + emittersField.setAccessible(true); + ((Map) emittersField.get(alertSseManager)).put(1L, deadEmitter); + + alertSseManager.broadcast("{\"id\":1}"); + + assertEquals(0, alertSseManager.subscriptionCount()); + assertNotNull(alertSseManager.createEmitter(2L)); + } } \ No newline at end of file diff --git a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java index 1fd681566c4..b721bd5df79 100644 --- a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java +++ b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java @@ -19,13 +19,16 @@ package org.apache.hertzbeat.manager.config; +import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.hertzbeat.common.constants.ManagerEventTypeEnum; import org.apache.hertzbeat.common.entity.dto.ImportTaskMessage; import org.apache.hertzbeat.common.entity.dto.ManagerMessage; import org.apache.hertzbeat.common.util.JsonUtil; +import org.springframework.http.HttpStatus; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; +import org.springframework.web.server.ResponseStatusException; import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @@ -40,16 +43,42 @@ @Slf4j @Component public class ManagerSseManager { + + /** + * How long a subscription may stay open before the client has to reconnect. + * + *

`Long.MAX_VALUE` meant a subscription never expired on its own, so a client that + * went away without closing cleanly held its request thread until the container noticed. + * A finite timeout bounds that; browsers reconnect on timeout, and the ui re-subscribes. + */ + private static final long EMITTER_TIMEOUT_MILLIS = 30 * 60 * 1000L; + + /** + * Cap on concurrently held subscriptions. Each one occupies a request thread, so without + * a ceiling enough parallel subscriptions exhaust the container's thread pool and take + * the whole application down with them. + */ + @Setter + private int maxEmitters = 1000; + private final Map emitters = new ConcurrentHashMap<>(); public SseEmitter createEmitter(Long clientId) { - SseEmitter emitter = new SseEmitter(Long.MAX_VALUE); + if (emitters.size() >= maxEmitters) { + log.warn("Refused manager subscription, already holding {} of at most {}", emitters.size(), maxEmitters); + throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "Too many manager subscriptions"); + } + SseEmitter emitter = new SseEmitter(EMITTER_TIMEOUT_MILLIS); emitter.onCompletion(() -> removeEmitter(clientId)); emitter.onTimeout(() -> removeEmitter(clientId)); emitters.put(clientId, emitter); return emitter; } + int subscriptionCount() { + return emitters.size(); + } + @Async public void broadcast(String eventName, String data) { emitters.forEach((clientId, emitter) -> { diff --git a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java new file mode 100644 index 00000000000..c09461dc573 --- /dev/null +++ b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java @@ -0,0 +1,65 @@ +/* + * 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.hertzbeat.manager.config; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpStatus; +import org.springframework.web.server.ResponseStatusException; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/** + * Test case for {@link ManagerSseManager}. + * + *

Every open subscription holds a request thread for as long as it lives, so both how + * long one may live and how many may exist at once have to be bounded. + */ +class ManagerSseManagerTest { + + private ManagerSseManager managerSseManager; + + @BeforeEach + void setUp() { + managerSseManager = new ManagerSseManager(); + } + + @Test + void testSubscriptionsAreGivenFiniteTimeout() { + SseEmitter emitter = managerSseManager.createEmitter(1L); + + assertNotNull(emitter.getTimeout()); + assertTrue(emitter.getTimeout() > 0 && emitter.getTimeout() < Long.MAX_VALUE, + "timeout must be finite, was " + emitter.getTimeout()); + } + + @Test + void testSubscriptionsBeyondLimitAreRefused() { + managerSseManager.setMaxEmitters(1); + + managerSseManager.createEmitter(1L); + ResponseStatusException thrown = + assertThrows(ResponseStatusException.class, () -> managerSseManager.createEmitter(2L)); + + assertEquals(HttpStatus.SERVICE_UNAVAILABLE, thrown.getStatusCode()); + assertEquals(1, managerSseManager.subscriptionCount()); + } +} diff --git a/hertzbeat-startup/src/main/resources/sureness.yml b/hertzbeat-startup/src/main/resources/sureness.yml index 0e8a8c975e5..5b45eeecea2 100644 --- a/hertzbeat-startup/src/main/resources/sureness.yml +++ b/hertzbeat-startup/src/main/resources/sureness.yml @@ -79,19 +79,21 @@ resourceRole: - /api/account/token===get===[admin] - /api/account/token/**===post===[admin] - /api/account/token/**===delete===[admin] + # the alert stream carries full alert payloads and the manager stream carries import + # progress; both are scoped like the log stream above rather than left anonymous + - /api/alert/sse/**===get===[admin,user,guest] + - /api/manager/sse/**===get===[admin,user,guest] # config the resource restful api that need bypass auth protection # rule: api===method # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - - /api/alert/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get - /api/observability/capability===get - /api/push/**===* - /api/status/page/public/**===* - - /api/manager/sse/**===* # web ui resource - /===get - /assets/**===get diff --git a/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java new file mode 100644 index 00000000000..041119b3a49 --- /dev/null +++ b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java @@ -0,0 +1,98 @@ +/* + * 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.hertzbeat.startup.security; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import com.usthe.sureness.matcher.util.TirePathTree; +import java.io.IOException; +import java.io.InputStream; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.yaml.snakeyaml.Yaml; + +/** + * Guards the rbac rules covering the server sent event streams. + * + *

`/api/alert/sse/**` and `/api/manager/sse/**` used to sit in `excludedResource`. + * Sureness evaluates the exclusion tree before any credential check, so an anonymous + * `curl -N` stayed subscribed and received every alert the deployment raised - internal + * hostnames, addresses, metric values and alert content - because + * `AlertNoticeDispatch` broadcasts each alert to every subscriber with no per subscriber + * filtering. `/api/logs/sse/**` was already scoped this way; these now match it. + */ +class SurenessSseRuleTest { + + private static final String SEPARATOR = "==="; + + private static TirePathTree roleTree; + + private static TirePathTree excludeTree; + + @BeforeAll + @SuppressWarnings("unchecked") + static void loadSurenessConfig() throws IOException { + List resourceRole; + List excludedResource; + try (InputStream in = SurenessSseRuleTest.class.getResourceAsStream("/sureness.yml")) { + assertNotNull(in, "sureness.yml must be on the classpath"); + Map document = new Yaml().load(in); + resourceRole = (List) document.get("resourceRole"); + excludedResource = (List) document.get("excludedResource"); + } + assertNotNull(resourceRole, "resourceRole must be present"); + assertNotNull(excludedResource, "excludedResource must be present"); + roleTree = new TirePathTree(); + roleTree.buildTree(new LinkedHashSet<>(resourceRole)); + excludeTree = new TirePathTree(); + excludeTree.buildTree(new LinkedHashSet<>(excludedResource)); + } + + @Test + void subscribingToAlertsRequiresAnAccount() { + assertEquals("[admin,user,guest]", + roleTree.searchPathFilterRoles("/api/alert/sse/subscribe" + SEPARATOR + "get")); + } + + @Test + void subscribingToManagerEventsRequiresAnAccount() { + assertEquals("[admin,user,guest]", + roleTree.searchPathFilterRoles("/api/manager/sse/subscribe" + SEPARATOR + "get")); + } + + /** + * The rule above only takes effect if the path stops matching an exclusion: sureness + * returns from `checkIn` as soon as `isExcludedResource` matches, before any credential + * is looked at. + */ + @Test + void theStreamsAreNoLongerAnonymous() { + assertNull(excludeTree.searchPathFilterRoles("/api/alert/sse/subscribe" + SEPARATOR + "get")); + assertNull(excludeTree.searchPathFilterRoles("/api/manager/sse/subscribe" + SEPARATOR + "get")); + } + + @Test + void theLogStreamScopingIsUnchanged() { + assertEquals("[admin,user,guest]", + roleTree.searchPathFilterRoles("/api/logs/sse/subscribe" + SEPARATOR + "get")); + } +} diff --git a/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml b/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml index ce2da6568ed..e121d64bbbe 100644 --- a/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml @@ -72,19 +72,21 @@ resourceRole: - /api/chat/**===post===[admin] - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] + # the alert stream carries full alert payloads and the manager stream carries import + # progress; both are scoped like the log stream above rather than left anonymous + - /api/alert/sse/**===get===[admin,user,guest] + - /api/manager/sse/**===get===[admin,user,guest] # config the resource restful api that need bypass auth protection # rule: api===method # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - - /api/alert/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get - /api/observability/capability===get - /api/push/**===* - /api/status/page/public/**===* - - /api/manager/sse/**===* # web ui resource - /===get - /assets/**===get diff --git a/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml b/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml index ce2da6568ed..e121d64bbbe 100644 --- a/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml @@ -72,19 +72,21 @@ resourceRole: - /api/chat/**===post===[admin] - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] + # the alert stream carries full alert payloads and the manager stream carries import + # progress; both are scoped like the log stream above rather than left anonymous + - /api/alert/sse/**===get===[admin,user,guest] + - /api/manager/sse/**===get===[admin,user,guest] # config the resource restful api that need bypass auth protection # rule: api===method # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - - /api/alert/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get - /api/observability/capability===get - /api/push/**===* - /api/status/page/public/**===* - - /api/manager/sse/**===* # web ui resource - /===get - /assets/**===get diff --git a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml index ce2da6568ed..e121d64bbbe 100644 --- a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml @@ -72,19 +72,21 @@ resourceRole: - /api/chat/**===post===[admin] - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] + # the alert stream carries full alert payloads and the manager stream carries import + # progress; both are scoped like the log stream above rather than left anonymous + - /api/alert/sse/**===get===[admin,user,guest] + - /api/manager/sse/**===get===[admin,user,guest] # config the resource restful api that need bypass auth protection # rule: api===method # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - - /api/alert/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get - /api/observability/capability===get - /api/push/**===* - /api/status/page/public/**===* - - /api/manager/sse/**===* # web ui resource - /===get - /assets/**===get diff --git a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml index 9b8736ba758..d63f3bf4198 100644 --- a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml @@ -76,19 +76,21 @@ resourceRole: - /api/ingestion/otlp/**===get===[admin,user,guest] - /api/logs/**===get===[admin,user,guest] - /api/traces/**===get===[admin,user,guest] + # the alert stream carries full alert payloads and the manager stream carries import + # progress; both are scoped like the log stream above rather than left anonymous + - /api/alert/sse/**===get===[admin,user,guest] + - /api/manager/sse/**===get===[admin,user,guest] # config the resource restful api that need bypass auth protection # rule: api===method # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - - /api/alert/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get - /api/observability/capability===get - /api/push/**===* - /api/status/page/public/**===* - - /api/manager/sse/**===* # web ui resource - /===get - /assets/**===get diff --git a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml index ce2da6568ed..e121d64bbbe 100644 --- a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml @@ -72,19 +72,21 @@ resourceRole: - /api/chat/**===post===[admin] - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] + # the alert stream carries full alert payloads and the manager stream carries import + # progress; both are scoped like the log stream above rather than left anonymous + - /api/alert/sse/**===get===[admin,user,guest] + - /api/manager/sse/**===get===[admin,user,guest] # config the resource restful api that need bypass auth protection # rule: api===method # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - - /api/alert/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get - /api/observability/capability===get - /api/push/**===* - /api/status/page/public/**===* - - /api/manager/sse/**===* # web ui resource - /===get - /assets/**===get diff --git a/script/sureness.yml b/script/sureness.yml index 5c38c9a8e30..f3bf2a47437 100644 --- a/script/sureness.yml +++ b/script/sureness.yml @@ -76,19 +76,21 @@ resourceRole: - /api/ingestion/otlp/**===get===[admin,user,guest] - /api/logs/**===get===[admin,user,guest] - /api/traces/**===get===[admin,user,guest] + # the alert stream carries full alert payloads and the manager stream carries import + # progress; both are scoped like the log stream above rather than left anonymous + - /api/alert/sse/**===get===[admin,user,guest] + - /api/manager/sse/**===get===[admin,user,guest] # config the resource restful api that need bypass auth protection # rule: api===method # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - - /api/alert/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get - /api/observability/capability===get - /api/push/**===* - /api/status/page/public/**===* - - /api/manager/sse/**===* # web ui resource - /===get - /assets/**===get diff --git a/web-app/src/app/layout/basic/widgets/notify.component.ts b/web-app/src/app/layout/basic/widgets/notify.component.ts index 022ff9a0e79..abb4b329b09 100644 --- a/web-app/src/app/layout/basic/widgets/notify.component.ts +++ b/web-app/src/app/layout/basic/widgets/notify.component.ts @@ -1,15 +1,17 @@ -import { ChangeDetectionStrategy, ChangeDetectorRef, Component, Inject, OnInit, OnDestroy } from '@angular/core'; +import { ChangeDetectionStrategy, ChangeDetectorRef, Component, Inject, NgZone, OnInit, OnDestroy } from '@angular/core'; import { Router } from '@angular/router'; import { I18NService } from '@core'; import { ALAIN_I18N_TOKEN } from '@delon/theme'; import { NzModalService } from 'ng-zorro-antd/modal'; import { NzNotificationService } from 'ng-zorro-antd/notification'; +import { Subscription } from 'rxjs'; import { finalize } from 'rxjs/operators'; import { Mute } from '../../../pojo/Mute'; import { SingleAlert } from '../../../pojo/SingleAlert'; import { AlertSoundService } from '../../../service/alert-sound.service'; import { AlertService } from '../../../service/alert.service'; +import { AuthorizedSseService } from '../../../service/authorized-sse.service'; import { GeneralConfigService } from '../../../service/general-config.service'; @Component({ @@ -125,7 +127,8 @@ export class HeaderNotifyComponent implements OnInit, OnDestroy { private previousCount = 0; // default to mute status mute: Mute = { mute: true }; - private eventSource!: EventSource; + private alertStream$!: Subscription; + private managerStream$!: Subscription; constructor( private router: Router, @Inject(ALAIN_I18N_TOKEN) private i18nSvc: I18NService, @@ -134,7 +137,9 @@ export class HeaderNotifyComponent implements OnInit, OnDestroy { private alertSvc: AlertService, private modal: NzModalService, private cdr: ChangeDetectorRef, - private alertSound: AlertSoundService + private alertSound: AlertSoundService, + private authorizedSseSvc: AuthorizedSseService, + private ngZone: NgZone ) {} ngOnInit(): void { @@ -166,9 +171,10 @@ export class HeaderNotifyComponent implements OnInit, OnDestroy { if (this.refreshInterval) { clearInterval(this.refreshInterval); } - if (this.eventSource) { - this.eventSource.close(); - } + // both streams are unsubscribed: they used to share one field, so the alert connection + // was never closed once the manager one overwrote it + this.alertStream$?.unsubscribe(); + this.managerStream$?.unsubscribe(); } onPopoverVisibleChange(visible: boolean): void { @@ -285,74 +291,70 @@ export class HeaderNotifyComponent implements OnInit, OnDestroy { } private initAlertSSEConnection(): void { - const sseUrl = '/api/alert/sse/subscribe'; - - this.eventSource = new EventSource(sseUrl); - - this.eventSource.addEventListener('ALERT_EVENT', (evt: MessageEvent) => { - let list: any[] = []; - let alert: SingleAlert = JSON.parse(evt.data); - let item = { - id: alert.id, - avatar: '/assets/img/notification.svg', - // title: `${alert.tags?.monitorName}--${this.i18nSvc.fanyi(`alert.severity.${alert.severity}`)}`, - title: alert.content, - datetime: new Date(alert.activeAt).toLocaleString(), - color: 'blue', - status: alert.status, - type: this.i18nSvc.fanyi('dashboard.alerts.title-no') - }; - console.log('alert:', alert); - list.push(item); - this.data = this.updateNoticeData(list); - if (!this.mute.mute && !this.notifiedAlert.includes(alert.id)) { - this.notifiedAlert.push(alert.id); - this.alertSound.playAlertSound(this.i18nSvc.currentLang); - const notification = new Notification(this.i18nSvc.fanyi('alert.notify.title'), { - body: this.i18nSvc.fanyi('alert.notify.body'), - icon: 'assets/logo.svg' - }); - notification.onclick = () => { - window.focus(); - this.router.navigateByUrl(`/alert/center`); - notification.close(); - }; - } - this.cdr.detectChanges(); + // read through AuthorizedSseService rather than EventSource: the stream now requires a + // credential, and EventSource cannot carry an Authorization header + this.alertStream$ = this.authorizedSseSvc.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({ + next: data => + this.ngZone.run(() => { + let list: any[] = []; + let alert: SingleAlert = JSON.parse(data); + let item = { + id: alert.id, + avatar: '/assets/img/notification.svg', + // title: `${alert.tags?.monitorName}--${this.i18nSvc.fanyi(`alert.severity.${alert.severity}`)}`, + title: alert.content, + datetime: new Date(alert.activeAt).toLocaleString(), + color: 'blue', + status: alert.status, + type: this.i18nSvc.fanyi('dashboard.alerts.title-no') + }; + console.log('alert:', alert); + list.push(item); + this.data = this.updateNoticeData(list); + if (!this.mute.mute && !this.notifiedAlert.includes(alert.id)) { + this.notifiedAlert.push(alert.id); + this.alertSound.playAlertSound(this.i18nSvc.currentLang); + const notification = new Notification(this.i18nSvc.fanyi('alert.notify.title'), { + body: this.i18nSvc.fanyi('alert.notify.body'), + icon: 'assets/logo.svg' + }); + notification.onclick = () => { + window.focus(); + this.router.navigateByUrl(`/alert/center`); + notification.close(); + }; + } + this.cdr.detectChanges(); + }), + error: error => console.error('SSE connection error:', error) }); - this.eventSource.onerror = error => { - console.error('SSE connection error:', error); - this.eventSource.close(); - }; } private initManagerSSEConnection(): void { - const sseUrl = '/api/manager/sse/subscribe'; - this.eventSource = new EventSource(sseUrl); - this.eventSource.addEventListener('IMPORT_TASK_EVENT', (evt: MessageEvent) => { - let msg = JSON.parse(evt.data); - if (msg.notifyLevel === 'SUCCESS') { - this.notifySvc.success( - this.i18nSvc.fanyi('common.notice'), - this.i18nSvc.fanyi('common.notify.import-success-detail', { taskName: msg.taskName }) - ); - } else if (msg.notifyLevel === 'ERROR') { - this.notifySvc.error( - this.i18nSvc.fanyi('common.notice'), - this.i18nSvc.fanyi('common.notify.import-fail-detail', { taskName: msg.taskName, errMsg: msg.errMsg }) - ); - } else if (msg.notifyLevel === 'INFO') { - this.notifySvc.info( - this.i18nSvc.fanyi('common.notice'), - this.i18nSvc.fanyi('common.notify.import-progress', { taskName: msg.taskName, progress: msg.progress }) - ); - } else { - console.error('Parse message error, msg:', evt.data); - } + this.managerStream$ = this.authorizedSseSvc.stream('/api/manager/sse/subscribe', 'IMPORT_TASK_EVENT').subscribe({ + next: data => + this.ngZone.run(() => { + let msg = JSON.parse(data); + if (msg.notifyLevel === 'SUCCESS') { + this.notifySvc.success( + this.i18nSvc.fanyi('common.notice'), + this.i18nSvc.fanyi('common.notify.import-success-detail', { taskName: msg.taskName }) + ); + } else if (msg.notifyLevel === 'ERROR') { + this.notifySvc.error( + this.i18nSvc.fanyi('common.notice'), + this.i18nSvc.fanyi('common.notify.import-fail-detail', { taskName: msg.taskName, errMsg: msg.errMsg }) + ); + } else if (msg.notifyLevel === 'INFO') { + this.notifySvc.info( + this.i18nSvc.fanyi('common.notice'), + this.i18nSvc.fanyi('common.notify.import-progress', { taskName: msg.taskName, progress: msg.progress }) + ); + } else { + console.error('Parse message error, msg:', data); + } + }), + error: error => console.error('Manager SSE connection error:', error) }); - this.eventSource.onerror = error => { - console.error('Manager SSE connection error:', error); - this.eventSource.close(); - }; } } diff --git a/web-app/src/app/routes/alert/alert-center/alert-center.component.ts b/web-app/src/app/routes/alert/alert-center/alert-center.component.ts index 2d734e89fba..6c94f3e1841 100644 --- a/web-app/src/app/routes/alert/alert-center/alert-center.component.ts +++ b/web-app/src/app/routes/alert/alert-center/alert-center.component.ts @@ -17,14 +17,16 @@ * under the License. */ -import { Component, Inject, OnDestroy, OnInit } from '@angular/core'; +import { Component, Inject, NgZone, OnDestroy, OnInit } from '@angular/core'; import { I18NService } from '@core'; import { ALAIN_I18N_TOKEN } from '@delon/theme'; import { NzModalService } from 'ng-zorro-antd/modal'; import { NzNotificationService } from 'ng-zorro-antd/notification'; +import { Subscription } from 'rxjs'; import { GroupAlert } from '../../../pojo/GroupAlert'; import { AlertService } from '../../../service/alert.service'; +import { AuthorizedSseService } from '../../../service/authorized-sse.service'; interface ExtendedGroupAlert extends GroupAlert { isNew?: boolean; @@ -39,6 +41,8 @@ export class AlertCenterComponent implements OnInit, OnDestroy { private notifySvc: NzNotificationService, private modal: NzModalService, private alertSvc: AlertService, + private authorizedSseSvc: AuthorizedSseService, + private ngZone: NgZone, @Inject(ALAIN_I18N_TOKEN) private i18nSvc: I18NService ) {} @@ -50,7 +54,7 @@ export class AlertCenterComponent implements OnInit, OnDestroy { checkedAlertIds = new Set(); filterStatus!: string; filterContent: string | undefined; - private eventSource!: EventSource; + private alertStream$!: Subscription; ngOnInit(): void { this.loadAlertsTable(); @@ -58,28 +62,24 @@ export class AlertCenterComponent implements OnInit, OnDestroy { } ngOnDestroy(): void { - if (this.eventSource) { - this.eventSource.close(); - } + this.alertStream$?.unsubscribe(); } // Initialize SSE subscription for real-time alerts private initSSESubscription(): void { - this.eventSource = new EventSource('/api/alert/sse/subscribe'); - this.eventSource.addEventListener('ALERT_EVENT', (evt: MessageEvent) => { - try { - const newAlert: GroupAlert = JSON.parse(evt.data); - this.updateAlertList(newAlert); - } catch (error) { - console.error('Error parsing SSE data:', error); - } + // read through AuthorizedSseService rather than EventSource: the stream now requires a + // credential, and EventSource cannot carry an Authorization header + this.alertStream$ = this.authorizedSseSvc.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({ + next: data => { + try { + const newAlert: GroupAlert = JSON.parse(data); + this.ngZone.run(() => this.updateAlertList(newAlert)); + } catch (error) { + console.error('Error parsing SSE data:', error); + } + }, + error: error => console.error('SSE connection error:', error) }); - - // Handle SSE errors - this.eventSource.onerror = error => { - console.error('SSE connection error:', error); - this.eventSource.close(); - }; } private updateAlertList(newAlert: GroupAlert): void { diff --git a/web-app/src/app/service/authorized-sse.service.spec.ts b/web-app/src/app/service/authorized-sse.service.spec.ts new file mode 100644 index 00000000000..895d67b7b3d --- /dev/null +++ b/web-app/src/app/service/authorized-sse.service.spec.ts @@ -0,0 +1,103 @@ +/* + * 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. + */ + +import { TestBed } from '@angular/core/testing'; + +import { AuthorizedSseService } from './authorized-sse.service'; +import { LocalStorageService } from './local-storage.service'; + +describe('AuthorizedSseService', () => { + let service: AuthorizedSseService; + let localStorageService: jasmine.SpyObj; + + /** Serves the given chunks as a readable body, the way a live sse response arrives. */ + function respondWith(chunks: string[], ok = true, status = 200): void { + const encoder = new TextEncoder(); + let index = 0; + const reader = { + read: () => (index < chunks.length ? Promise.resolve({ value: encoder.encode(chunks[index++]), done: false }) : new Promise(() => {})) // an open stream never completes on its own + }; + spyOn(window, 'fetch').and.returnValue(Promise.resolve({ ok, status, body: { getReader: () => reader } } as any)); + } + + beforeEach(() => { + localStorageService = jasmine.createSpyObj('LocalStorageService', ['getAuthorizationToken']); + TestBed.configureTestingModule({ + providers: [{ provide: LocalStorageService, useValue: localStorageService }] + }); + service = TestBed.inject(AuthorizedSseService); + }); + + it('sends the stored token so the stream can require a credential', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith(['event:ALERT_EVENT\ndata:{"id":1}\n\n']); + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(() => { + const [, init] = (window.fetch as jasmine.Spy).calls.mostRecent().args; + expect((init.headers as Record)['Authorization']).toBe('Bearer a-token'); + subscription.unsubscribe(); + done(); + }); + }); + + it('emits the data payload of a matching event', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith(['event:ALERT_EVENT\ndata:{"id":1}\n\n']); + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(data => { + expect(data).toBe('{"id":1}'); + subscription.unsubscribe(); + done(); + }); + }); + + it('ignores events of another name on the same stream', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith(['event:OTHER_EVENT\ndata:{"id":1}\n\n', 'event:ALERT_EVENT\ndata:{"id":2}\n\n']); + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(data => { + expect(data).toBe('{"id":2}'); + subscription.unsubscribe(); + done(); + }); + }); + + it('reassembles an event split across chunks', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith(['event:ALERT_EVENT\ndata:{"id"', ':3}\n\n']); + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(data => { + expect(data).toBe('{"id":3}'); + subscription.unsubscribe(); + done(); + }); + }); + + it('surfaces a rejected subscription as an error', done => { + localStorageService.getAuthorizationToken.and.returnValue(null as any); + respondWith([], false, 401); + + service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({ + error: error => { + expect(String(error)).toContain('401'); + done(); + } + }); + }); +}); diff --git a/web-app/src/app/service/authorized-sse.service.ts b/web-app/src/app/service/authorized-sse.service.ts new file mode 100644 index 00000000000..2be4178ba1d --- /dev/null +++ b/web-app/src/app/service/authorized-sse.service.ts @@ -0,0 +1,109 @@ +/* + * 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. + */ + +import { Injectable, NgZone } from '@angular/core'; +import { Observable } from 'rxjs'; + +import { LocalStorageService } from './local-storage.service'; + +/** + * Reads a server sent event stream with the bearer token attached. + * + * The browser's own `EventSource` cannot carry an `Authorization` header, which is why the + * alert and manager streams used to be reachable without any credential at all. Reading the + * stream through `fetch` instead lets the token travel with the request, so the endpoints can + * be moved behind the same rbac rules as the rest of the api. + * + * The returned observable starts the request on subscribe and aborts it on unsubscribe. + * Events are emitted outside the angular zone; a caller that touches component state should + * re-enter the zone itself, as it would with any other stream. + */ +@Injectable({ providedIn: 'root' }) +export class AuthorizedSseService { + constructor(private localStorageService: LocalStorageService, private ngZone: NgZone) {} + + /** + * @param url stream endpoint, relative to the origin + * @param eventName name of the sse event to emit; other events are ignored + * @returns the `data` payload of every matching event, as raw text + */ + stream(url: string, eventName: string): Observable { + return new Observable(subscriber => { + const abortController = new AbortController(); + const token = this.localStorageService.getAuthorizationToken(); + const headers: Record = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache' + }; + if (token) { + headers['Authorization'] = `Bearer ${token}`; + } + + this.ngZone.runOutsideAngular(() => { + fetch(url, { method: 'GET', headers, signal: abortController.signal }) + .then(async response => { + if (!response.ok) { + throw new Error(`SSE request to ${url} failed with status ${response.status}`); + } + const reader = response.body?.getReader(); + if (!reader) { + throw new Error(`SSE response from ${url} has no readable body`); + } + + const decoder = new TextDecoder(); + let buffer = ''; + while (!abortController.signal.aborted) { + const { value, done } = await reader.read(); + if (done) { + throw new Error(`SSE connection to ${url} closed`); + } + + buffer += decoder.decode(value, { stream: true }); + const frames = buffer.split(/\r?\n\r?\n/); + buffer = frames.pop() ?? ''; + + for (const frame of frames) { + let frameEvent = ''; + const dataLines: string[] = []; + for (const line of frame.split(/\r?\n/)) { + if (line.startsWith('event:')) { + frameEvent = line.substring(6).trim(); + } else if (line.startsWith('data:')) { + dataLines.push(line.substring(5)); + } + } + if (frameEvent !== eventName || dataLines.length === 0) { + continue; + } + subscriber.next(dataLines.join('\n')); + } + } + }) + .catch(error => { + if (abortController.signal.aborted) { + return; + } + subscriber.error(error); + }); + }); + + return () => abortController.abort(); + }); + } +} From 5ea6d9461572790c4fe1d8f51381fc44b8d0d1a7 Mon Sep 17 00:00:00 2001 From: Duansg Date: Tue, 4 Aug 2026 08:04:14 -0700 Subject: [PATCH 2/2] [fix] reconnect the authorized sse stream instead of giving up The alert and manager streams moved from `EventSource` to a `fetch` reader so the bearer token could travel with the request. `EventSource` reconnects on its own; the reader did not, and the same change bounds an emitter's life to thirty minutes. Together that meant the notification bell and the alert center stopped receiving anything half an hour in, until the page was reloaded. `AuthorizedSseService` now re-establishes a dropped stream itself, with an exponential backoff from one second to thirty, reset once a connection is up. The token is read on every attempt so a refreshed one is picked up. A 401 or 403 still ends the observable: reconnecting cannot change that answer and retrying would only hammer the endpoint. This mirrors `log-stream.component.ts`, which already reads an authorized stream through `fetch` and already reconnects. Co-Authored-By: Claude Opus 5 (1M context) --- .../alert/config/AlertSseManager.java | 4 +- .../manager/config/ManagerSseManager.java | 4 +- .../service/authorized-sse.service.spec.ts | 110 ++++++++++++ .../src/app/service/authorized-sse.service.ts | 157 ++++++++++++------ 4 files changed, 224 insertions(+), 51 deletions(-) diff --git a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java index 461846b5800..3434c7c5997 100644 --- a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java +++ b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java @@ -45,7 +45,9 @@ public class AlertSseManager { * *

`Long.MAX_VALUE` meant a subscription never expired on its own, so a client that * went away without closing cleanly held its request thread until the container noticed. - * A finite timeout bounds that; browsers reconnect on timeout, and the ui re-subscribes. + * A finite timeout bounds that. The timeout is only safe because the ui reconnects when + * the stream ends: `AuthorizedSseService` reads through `fetch` rather than + * `EventSource`, so it has to reconnect itself, and it does. */ private static final long EMITTER_TIMEOUT_MILLIS = 30 * 60 * 1000L; diff --git a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java index b721bd5df79..d9e4b90483f 100644 --- a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java +++ b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java @@ -49,7 +49,9 @@ public class ManagerSseManager { * *

`Long.MAX_VALUE` meant a subscription never expired on its own, so a client that * went away without closing cleanly held its request thread until the container noticed. - * A finite timeout bounds that; browsers reconnect on timeout, and the ui re-subscribes. + * A finite timeout bounds that. The timeout is only safe because the ui reconnects when + * the stream ends: `AuthorizedSseService` reads through `fetch` rather than + * `EventSource`, so it has to reconnect itself, and it does. */ private static final long EMITTER_TIMEOUT_MILLIS = 30 * 60 * 1000L; diff --git a/web-app/src/app/service/authorized-sse.service.spec.ts b/web-app/src/app/service/authorized-sse.service.spec.ts index 895d67b7b3d..77d13032323 100644 --- a/web-app/src/app/service/authorized-sse.service.spec.ts +++ b/web-app/src/app/service/authorized-sse.service.spec.ts @@ -36,6 +36,31 @@ describe('AuthorizedSseService', () => { spyOn(window, 'fetch').and.returnValue(Promise.resolve({ ok, status, body: { getReader: () => reader } } as any)); } + /** + * Serves one response per connection attempt, so a test can let the first one end and + * assert on what the service does next. A `null` entry stands for a request that fails + * outright rather than returning a response. + */ + function respondPerAttempt(attempts: Array): void { + const encoder = new TextEncoder(); + let attempt = 0; + spyOn(window, 'fetch').and.callFake(() => { + const chunks = attempts[Math.min(attempt++, attempts.length - 1)]; + if (chunks === null) { + return Promise.reject(new Error('network down')); + } + let index = 0; + const reader = { + // once the chunks run out the server closes the stream, as an emitter timeout does + read: () => + index < chunks.length + ? Promise.resolve({ value: encoder.encode(chunks[index++]), done: false }) + : Promise.resolve({ value: undefined, done: true }) + }; + return Promise.resolve({ ok: true, status: 200, body: { getReader: () => reader } } as any); + }); + } + beforeEach(() => { localStorageService = jasmine.createSpyObj('LocalStorageService', ['getAuthorizationToken']); TestBed.configureTestingModule({ @@ -100,4 +125,89 @@ describe('AuthorizedSseService', () => { } }); }); + + /** + * The server closes a subscription once its emitter times out, so a stream that gave up + * at the first close would stop delivering alerts half an hour in. `EventSource` used to + * reconnect on its own; reading through `fetch` has to do it here instead. + */ + it('reconnects after the server closes the stream', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondPerAttempt([['event:ALERT_EVENT\ndata:{"id":1}\n\n'], ['event:ALERT_EVENT\ndata:{"id":2}\n\n']]); + service.initialRetryDelayMillis = 0; + + const received: string[] = []; + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(data => { + received.push(data); + if (received.length === 2) { + expect(received).toEqual(['{"id":1}', '{"id":2}']); + expect((window.fetch as jasmine.Spy).calls.count()).toBe(2); + subscription.unsubscribe(); + done(); + } + }); + }); + + it('reconnects after a failed connection attempt', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondPerAttempt([null, ['event:ALERT_EVENT\ndata:{"id":9}\n\n']]); + service.initialRetryDelayMillis = 0; + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(data => { + expect(data).toBe('{"id":9}'); + expect((window.fetch as jasmine.Spy).calls.count()).toBe(2); + subscription.unsubscribe(); + done(); + }); + }); + + it('re-reads the token on every attempt so a refreshed one is used', done => { + localStorageService.getAuthorizationToken.and.returnValues('stale-token', 'fresh-token'); + respondPerAttempt([['event:ALERT_EVENT\ndata:{"id":1}\n\n'], ['event:ALERT_EVENT\ndata:{"id":2}\n\n']]); + service.initialRetryDelayMillis = 0; + + let count = 0; + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(() => { + if (++count === 2) { + const [, init] = (window.fetch as jasmine.Spy).calls.mostRecent().args; + expect((init.headers as Record)['Authorization']).toBe('Bearer fresh-token'); + subscription.unsubscribe(); + done(); + } + }); + }); + + it('stops reconnecting once the subscription is closed', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondPerAttempt([['event:ALERT_EVENT\ndata:{"id":1}\n\n']]); + service.initialRetryDelayMillis = 0; + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(() => { + subscription.unsubscribe(); + const attemptsAtUnsubscribe = (window.fetch as jasmine.Spy).calls.count(); + setTimeout(() => { + expect((window.fetch as jasmine.Spy).calls.count()).toBe(attemptsAtUnsubscribe); + done(); + }, 20); + }); + }); + + /** + * A credential the server refuses is the one failure reconnecting cannot fix, so it has to + * end the observable instead of turning into a retry loop against the endpoint. + */ + it('does not retry a rejected credential', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith([], false, 403); + service.initialRetryDelayMillis = 0; + + service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({ + error: () => { + setTimeout(() => { + expect((window.fetch as jasmine.Spy).calls.count()).toBe(1); + done(); + }, 20); + } + }); + }); }); diff --git a/web-app/src/app/service/authorized-sse.service.ts b/web-app/src/app/service/authorized-sse.service.ts index 2be4178ba1d..02e3f659ecf 100644 --- a/web-app/src/app/service/authorized-sse.service.ts +++ b/web-app/src/app/service/authorized-sse.service.ts @@ -33,9 +33,25 @@ import { LocalStorageService } from './local-storage.service'; * The returned observable starts the request on subscribe and aborts it on unsubscribe. * Events are emitted outside the angular zone; a caller that touches component state should * re-enter the zone itself, as it would with any other stream. + * + * A dropped connection is re-established rather than surfaced as an error, because the + * server closes an idle subscription on purpose once its emitter times out. `EventSource` + * used to reconnect on its own, so reading through `fetch` has to carry that behaviour over + * or a stream would simply stop delivering. Only a rejected credential ends the observable: + * reconnecting cannot change that answer, and retrying would hammer the endpoint. */ @Injectable({ providedIn: 'root' }) export class AuthorizedSseService { + /** + * Delay before the first reconnect attempt; doubles on each consecutive failure. + * An instance field rather than a constant so a test can drive the reconnect without + * waiting a real second. + */ + initialRetryDelayMillis = 1000; + + /** Ceiling for the doubling above, so a server that stays down is polled at a fixed rate. */ + maxRetryDelayMillis = 30000; + constructor(private localStorageService: LocalStorageService, private ngZone: NgZone) {} /** @@ -45,65 +61,108 @@ export class AuthorizedSseService { */ stream(url: string, eventName: string): Observable { return new Observable(subscriber => { - const abortController = new AbortController(); - const token = this.localStorageService.getAuthorizationToken(); - const headers: Record = { - Accept: 'text/event-stream', - 'Cache-Control': 'no-cache' + let controller: AbortController | undefined; + let retryTimer: ReturnType | undefined; + let retryDelay = this.initialRetryDelayMillis; + let stopped = false; + + const scheduleReconnect = (): void => { + if (stopped) { + return; + } + const delay = retryDelay; + retryDelay = Math.min(retryDelay * 2, this.maxRetryDelayMillis); + retryTimer = setTimeout(connect, delay); }; - if (token) { - headers['Authorization'] = `Bearer ${token}`; - } - this.ngZone.runOutsideAngular(() => { - fetch(url, { method: 'GET', headers, signal: abortController.signal }) - .then(async response => { - if (!response.ok) { - throw new Error(`SSE request to ${url} failed with status ${response.status}`); - } - const reader = response.body?.getReader(); - if (!reader) { - throw new Error(`SSE response from ${url} has no readable body`); - } + const connect = (): void => { + if (stopped) { + return; + } + const current = new AbortController(); + controller = current; + // read the token on every attempt: it may have been refreshed since the last one + const token = this.localStorageService.getAuthorizationToken(); + const headers: Record = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache' + }; + if (token) { + headers['Authorization'] = `Bearer ${token}`; + } - const decoder = new TextDecoder(); - let buffer = ''; - while (!abortController.signal.aborted) { - const { value, done } = await reader.read(); - if (done) { - throw new Error(`SSE connection to ${url} closed`); + this.ngZone.runOutsideAngular(() => { + fetch(url, { method: 'GET', headers, signal: current.signal }) + .then(async response => { + if (response.status === 401 || response.status === 403) { + stopped = true; + subscriber.error(new Error(`SSE request to ${url} failed with status ${response.status}`)); + return; + } + if (!response.ok) { + throw new Error(`SSE request to ${url} failed with status ${response.status}`); } + const reader = response.body?.getReader(); + if (!reader) { + throw new Error(`SSE response from ${url} has no readable body`); + } + // the stream is up, so a later drop starts its backoff from the bottom again + retryDelay = this.initialRetryDelayMillis; - buffer += decoder.decode(value, { stream: true }); - const frames = buffer.split(/\r?\n\r?\n/); - buffer = frames.pop() ?? ''; + const decoder = new TextDecoder(); + let buffer = ''; + while (!current.signal.aborted) { + const { value, done } = await reader.read(); + if (done) { + // the server closed it, most likely an emitter timeout; reconnect below + return; + } - for (const frame of frames) { - let frameEvent = ''; - const dataLines: string[] = []; - for (const line of frame.split(/\r?\n/)) { - if (line.startsWith('event:')) { - frameEvent = line.substring(6).trim(); - } else if (line.startsWith('data:')) { - dataLines.push(line.substring(5)); + buffer += decoder.decode(value, { stream: true }); + const frames = buffer.split(/\r?\n\r?\n/); + buffer = frames.pop() ?? ''; + + for (const frame of frames) { + let frameEvent = ''; + const dataLines: string[] = []; + for (const line of frame.split(/\r?\n/)) { + if (line.startsWith('event:')) { + frameEvent = line.substring(6).trim(); + } else if (line.startsWith('data:')) { + dataLines.push(line.substring(5)); + } } + if (frameEvent !== eventName || dataLines.length === 0) { + continue; + } + subscriber.next(dataLines.join('\n')); } - if (frameEvent !== eventName || dataLines.length === 0) { - continue; - } - subscriber.next(dataLines.join('\n')); } - } - }) - .catch(error => { - if (abortController.signal.aborted) { - return; - } - subscriber.error(error); - }); - }); + }) + .then(() => { + if (!stopped && !current.signal.aborted) { + scheduleReconnect(); + } + }) + .catch(error => { + if (stopped || current.signal.aborted) { + return; + } + console.error(`SSE connection to ${url} interrupted, reconnecting`, error); + scheduleReconnect(); + }); + }); + }; - return () => abortController.abort(); + connect(); + + return () => { + stopped = true; + if (retryTimer) { + clearTimeout(retryTimer); + } + controller?.abort(); + }; }); } }