From c8f2da472e02de4a687ce4fde8a665ffca7be965 Mon Sep 17 00:00:00 2001 From: Yang Chen <1597081640@qq.com> Date: Sat, 25 Jul 2026 13:09:13 +0800 Subject: [PATCH] fix: add SSE authentication and correct GreptimeDB SQL (#4243) --- .gitignore | 4 + .../src/test/resources/sureness.yml | 2 +- .../greptime/tables/hertzbeat_logs.sql | 2 +- .../resources/greptime/tables/hzb_traces.sql | 2 +- .../src/test/resources/sureness.yml | 2 +- .../src/main/resources/sureness.yml | 2 +- .../hertzbeat-mysql-iotdb/conf/sureness.yml | 2 +- .../conf/sureness.yml | 2 +- .../conf/sureness.yml | 2 +- .../conf/sureness.yml | 2 +- .../conf/sureness.yml | 2 +- script/sureness.yml | 2 +- .../log-stream/log-stream.component.spec.ts | 29 +++- .../log/log-stream/log-stream.component.ts | 148 ++++++++++++------ 14 files changed, 145 insertions(+), 58 deletions(-) diff --git a/.gitignore b/.gitignore index 0c6673035e..b3611b7620 100644 --- a/.gitignore +++ b/.gitignore @@ -56,3 +56,7 @@ application-mysql.yml application-pg.yml /package.json /yarn.lock + +# Local Codex workspace files +.codex/ +AGENTS.md diff --git a/hertzbeat-e2e/hertzbeat-log-e2e/src/test/resources/sureness.yml b/hertzbeat-e2e/hertzbeat-log-e2e/src/test/resources/sureness.yml index cdda55e93a..e7784d87ca 100644 --- a/hertzbeat-e2e/hertzbeat-log-e2e/src/test/resources/sureness.yml +++ b/hertzbeat-e2e/hertzbeat-log-e2e/src/test/resources/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/sse/**===post===[admin,user] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin,user] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] - /api/otlp/**===post===[admin,user] - /api/ingestion/otlp/**===get===[admin,user,guest] @@ -81,7 +82,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get diff --git a/hertzbeat-log/src/main/resources/greptime/tables/hertzbeat_logs.sql b/hertzbeat-log/src/main/resources/greptime/tables/hertzbeat_logs.sql index ea5c4d25f1..a9d7d0dc56 100644 --- a/hertzbeat-log/src/main/resources/greptime/tables/hertzbeat_logs.sql +++ b/hertzbeat-log/src/main/resources/greptime/tables/hertzbeat_logs.sql @@ -29,4 +29,4 @@ CREATE TABLE IF NOT EXISTS hertzbeat_logs ( "resource" JSON NULL, "instrumentation_scope" JSON NULL, "dropped_attributes_count" INT NULL -) WITH (append_mode = true); +) WITH ('append_mode' = 'true'); diff --git a/hertzbeat-log/src/main/resources/greptime/tables/hzb_traces.sql b/hertzbeat-log/src/main/resources/greptime/tables/hzb_traces.sql index 64585ce31b..c0d62a0a44 100644 --- a/hertzbeat-log/src/main/resources/greptime/tables/hzb_traces.sql +++ b/hertzbeat-log/src/main/resources/greptime/tables/hzb_traces.sql @@ -35,4 +35,4 @@ CREATE TABLE IF NOT EXISTS hzb_traces ( "span_events" JSON NULL, "span_links" JSON NULL, PRIMARY KEY("service_name") -) WITH (append_mode = true, table_data_model = 'greptime_trace_v1'); +) WITH ('append_mode' = 'true', 'table_data_model' = 'greptime_trace_v1'); diff --git a/hertzbeat-manager/src/test/resources/sureness.yml b/hertzbeat-manager/src/test/resources/sureness.yml index cdda55e93a..e7784d87ca 100644 --- a/hertzbeat-manager/src/test/resources/sureness.yml +++ b/hertzbeat-manager/src/test/resources/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/sse/**===post===[admin,user] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin,user] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] - /api/otlp/**===post===[admin,user] - /api/ingestion/otlp/**===get===[admin,user,guest] @@ -81,7 +82,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get diff --git a/hertzbeat-startup/src/main/resources/sureness.yml b/hertzbeat-startup/src/main/resources/sureness.yml index 15e2a2b102..0e8a8c975e 100644 --- a/hertzbeat-startup/src/main/resources/sureness.yml +++ b/hertzbeat-startup/src/main/resources/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/mcp/**===post===[admin] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] - /api/otlp/**===post===[admin,user] - /api/ingestion/otlp/**===get===[admin,user,guest] @@ -84,7 +85,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get diff --git a/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml b/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml index 70609bf26e..ce2da6568e 100644 --- a/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/mcp/**===post===[admin] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] # config the resource restful api that need bypass auth protection @@ -77,7 +78,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get diff --git a/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml b/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml index 70609bf26e..ce2da6568e 100644 --- a/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/mcp/**===post===[admin] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] # config the resource restful api that need bypass auth protection @@ -77,7 +78,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===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 70609bf26e..ce2da6568e 100644 --- a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/mcp/**===post===[admin] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] # config the resource restful api that need bypass auth protection @@ -77,7 +78,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get diff --git a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml index edc94a4ef4..9b8736ba75 100644 --- a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/mcp/**===post===[admin] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] - /api/otlp/**===post===[admin,user] - /api/ingestion/otlp/**===get===[admin,user,guest] @@ -81,7 +82,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===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 70609bf26e..ce2da6568e 100644 --- a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/mcp/**===post===[admin] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] # config the resource restful api that need bypass auth protection @@ -77,7 +78,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get diff --git a/script/sureness.yml b/script/sureness.yml index 47b0270e4c..5c38c9a8e3 100644 --- a/script/sureness.yml +++ b/script/sureness.yml @@ -70,6 +70,7 @@ resourceRole: - /api/mcp/**===post===[admin] - /api/chat/**===get===[admin,user] - /api/chat/**===post===[admin] + - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] - /api/otlp/**===post===[admin,user] - /api/ingestion/otlp/**===get===[admin,user,guest] @@ -81,7 +82,6 @@ resourceRole: # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth. excludedResource: - /api/alert/sse/**===* - - /api/logs/sse/**===* - /api/account/auth/**===* - /api/i18n/**===get - /api/apps/hierarchy===get diff --git a/web-app/src/app/routes/log/log-stream/log-stream.component.spec.ts b/web-app/src/app/routes/log/log-stream/log-stream.component.spec.ts index 393c1c9310..b3e6908e32 100644 --- a/web-app/src/app/routes/log/log-stream/log-stream.component.spec.ts +++ b/web-app/src/app/routes/log/log-stream/log-stream.component.spec.ts @@ -24,18 +24,45 @@ import { LogStreamComponent } from './log-stream.component'; describe('LogStreamComponent', () => { let component: LogStreamComponent; let fixture: ComponentFixture; + let fetchSpy: jasmine.Spy; beforeEach(async () => { + localStorage.setItem('Authorization', 'test-token'); + fetchSpy = spyOn(window, 'fetch').and.returnValue(new Promise(() => {})); + await TestBed.configureTestingModule({ imports: [LogStreamComponent] - }).compileComponents(); + }) + .overrideComponent(LogStreamComponent, { + set: { template: '' } + }) + .compileComponents(); fixture = TestBed.createComponent(LogStreamComponent); component = fixture.componentInstance; fixture.detectChanges(); }); + afterEach(() => { + fixture.destroy(); + localStorage.removeItem('Authorization'); + }); + it('should create', () => { expect(component).toBeTruthy(); }); + + it('should authenticate the log stream request', () => { + expect(fetchSpy).toHaveBeenCalledWith( + '/api/logs/sse/subscribe', + jasmine.objectContaining({ + method: 'GET', + headers: jasmine.objectContaining({ + Accept: 'text/event-stream', + Authorization: 'Bearer test-token' + }), + signal: jasmine.any(AbortSignal) + }) + ); + }); }); diff --git a/web-app/src/app/routes/log/log-stream/log-stream.component.ts b/web-app/src/app/routes/log/log-stream/log-stream.component.ts index e5d7c1093e..c1c0393712 100644 --- a/web-app/src/app/routes/log/log-stream/log-stream.component.ts +++ b/web-app/src/app/routes/log/log-stream/log-stream.component.ts @@ -49,6 +49,7 @@ import { NzTagModule } from 'ng-zorro-antd/tag'; import { NzToolTipModule } from 'ng-zorro-antd/tooltip'; import { LogEntry } from '../../../pojo/LogEntry'; +import { LocalStorageService } from '../../../service/local-storage.service'; interface ExtendedLogEntry { original: LogEntry; @@ -85,7 +86,8 @@ interface ExtendedLogEntry { export class LogStreamComponent implements OnInit, OnDestroy, AfterViewInit { @Input() embedded = false; // SSE connection and state - private eventSource!: EventSource; + private streamAbortController?: AbortController; + private reconnectTimer?: ReturnType; isConnected: boolean = false; isConnecting: boolean = false; @@ -124,7 +126,12 @@ export class LogStreamComponent implements OnInit, OnDestroy, AfterViewInit { // ViewChild for log container @ViewChild(CdkVirtualScrollViewport) viewport!: CdkVirtualScrollViewport; - constructor(@Inject(ALAIN_I18N_TOKEN) private i18nSvc: I18NService, private cdr: ChangeDetectorRef, private ngZone: NgZone) {} + constructor( + @Inject(ALAIN_I18N_TOKEN) private i18nSvc: I18NService, + private cdr: ChangeDetectorRef, + private ngZone: NgZone, + private localStorageService: LocalStorageService + ) {} ngOnInit(): void { this.connectToLogStream(); @@ -148,7 +155,7 @@ export class LogStreamComponent implements OnInit, OnDestroy, AfterViewInit { } private connectToLogStream(): void { - if (this.eventSource) { + if (this.streamAbortController) { this.disconnectFromLogStream(); } @@ -157,59 +164,108 @@ export class LogStreamComponent implements OnInit, OnDestroy, AfterViewInit { // Build filter parameters const filterParams = this.buildFilterParams(); const url = `/api/logs/sse/subscribe${filterParams ? `?${filterParams}` : ''}`; + const token = this.localStorageService.getAuthorizationToken(); + const headers: Record = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache' + }; + if (token) { + headers['Authorization'] = `Bearer ${token}`; + } + const abortController = new AbortController(); + this.streamAbortController = abortController; - try { - this.eventSource = new EventSource(url); + this.ngZone.runOutsideAngular(() => { + fetch(url, { + method: 'GET', + headers, + signal: abortController.signal + }) + .then(async response => { + if (!response.ok) { + throw new Error(`Log stream request failed with status ${response.status}`); + } + const reader = response.body?.getReader(); + if (!reader) { + throw new Error('Log stream response has no readable body'); + } - this.eventSource.onopen = () => { - this.ngZone.run(() => { - this.isConnected = true; - this.isConnecting = false; - this.cdr.markForCheck(); - }); - }; + const decoder = new TextDecoder(); + let buffer = ''; + this.ngZone.run(() => { + this.isConnected = true; + this.isConnecting = false; + this.cdr.markForCheck(); + }); - // Run outside Angular zone to prevent change detection on every message - this.ngZone.runOutsideAngular(() => { - this.eventSource.addEventListener('LOG_EVENT', (evt: MessageEvent) => { - if (!this.isPaused) { - try { - const logEntry: LogEntry = JSON.parse(evt.data); - this.queueLogEntry(logEntry); - } catch (error) { - // Silently ignore parse errors in high TPS scenario - console.error(error); + while (!abortController.signal.aborted) { + const { value, done } = await reader.read(); + if (done) { + throw new Error('Log stream connection closed'); + } + + buffer += decoder.decode(value, { stream: true }); + const frames = buffer.split(/\r?\n\r?\n/); + buffer = frames.pop() ?? ''; + + for (const frame of frames) { + let eventType = ''; + const dataLines: string[] = []; + for (const line of frame.split(/\r?\n/)) { + if (line.startsWith('event:')) { + eventType = line.substring(6); + } else if (line.startsWith('data:')) { + dataLines.push(line.substring(5)); + } + } + + if (eventType !== 'LOG_EVENT' || dataLines.length === 0 || this.isPaused) { + continue; + } + + try { + const logEntry: LogEntry = JSON.parse(dataLines.join('\n')); + this.queueLogEntry(logEntry); + } catch (error) { + // A malformed event must not terminate a long-lived log stream. + console.error('Failed to parse log stream event:', error); + } } } - }); - }); - - this.eventSource.onerror = error => { - this.ngZone.run(() => { - this.isConnected = false; - this.isConnecting = false; - this.cdr.markForCheck(); - }); - - // Auto-reconnect after 5 seconds - setTimeout(() => { - if (!this.isConnected) { - this.connectToLogStream(); + }) + .catch(error => { + if (abortController.signal.aborted) { + return; } - }, 5000); - }; - } catch (error) { - this.isConnecting = false; - console.error('Failed to create EventSource:', error); - } + + console.error('Log stream connection error:', error); + this.ngZone.run(() => { + this.isConnected = false; + this.isConnecting = false; + this.cdr.markForCheck(); + }); + + this.reconnectTimer = setTimeout(() => { + if (!this.isConnected && this.streamAbortController === abortController) { + this.streamAbortController = undefined; + this.connectToLogStream(); + } + }, 5000); + }); + }); } private disconnectFromLogStream(): void { - if (this.eventSource) { - this.eventSource.close(); - this.isConnected = false; - this.isConnecting = false; + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = undefined; } + if (this.streamAbortController) { + this.streamAbortController.abort(); + this.streamAbortController = undefined; + } + this.isConnected = false; + this.isConnecting = false; } private buildFilterParams(): string {