You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ta...@apache.org on 2024/01/02 13:48:32 UTC
(flink) 01/02: [FLINK-33433][rest] Introduce async-profiler to support profiling JobManager via REST API
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 240494fd6169cb98b47808a003ee00804a780360
Author: Yu Chen <yu...@gmail.com>
AuthorDate: Mon Nov 6 03:50:18 2023 +0800
[FLINK-33433][rest] Introduce async-profiler to support profiling JobManager via REST API
---
.../shortcodes/generated/expert_rest_section.html | 24 +++
.../shortcodes/generated/rest_configuration.html | 24 +++
.../apache/flink/configuration/RestOptions.java | 37 ++++
flink-dist/src/main/resources/META-INF/NOTICE | 1 +
.../src/app/components/humanize-watermark.pipe.ts | 4 +-
.../src/app/interfaces/job-profiler.ts | 31 +++
.../app/pages/job-manager/job-manager.component.ts | 3 +-
.../profiler/job-manager-profiler.component.html | 109 ++++++++++
.../profiler/job-manager-profiler.component.less | 43 ++++
.../profiler/job-manager-profiler.component.ts | 151 +++++++++++++
.../src/app/pages/job-manager/routes.ts | 8 +
.../src/app/services/job-manager.service.ts | 24 +++
flink-runtime/pom.xml | 7 +
.../cluster/JobManagerProfilingFileHandler.java | 63 ++++++
.../cluster/JobManagerProfilingHandler.java | 77 +++++++
.../cluster/JobManagerProfilingListHandler.java | 63 ++++++
.../messages/ProfilingFileNamePathParameter.java | 46 ++++
.../flink/runtime/rest/messages/ProfilingInfo.java | 187 ++++++++++++++++
.../runtime/rest/messages/ProfilingInfoList.java | 79 +++++++
.../cluster/JobManagerProfilingFileHeaders.java | 63 ++++++
.../cluster/JobManagerProfilingHeaders.java | 87 ++++++++
.../cluster/JobManagerProfilingListHeaders.java | 89 ++++++++
.../cluster/ProfilingFileMessageParameters.java | 45 ++++
.../messages/cluster/ProfilingRequestBody.java | 57 +++++
.../runtime/util/profiler/ProfilingService.java | 237 +++++++++++++++++++++
.../runtime/webmonitor/WebMonitorEndpoint.java | 42 ++++
26 files changed, 1598 insertions(+), 3 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/expert_rest_section.html b/docs/layouts/shortcodes/generated/expert_rest_section.html
index 6c0d5e1a643..4c7d98d45ed 100644
--- a/docs/layouts/shortcodes/generated/expert_rest_section.html
+++ b/docs/layouts/shortcodes/generated/expert_rest_section.html
@@ -86,6 +86,30 @@
<td>Long</td>
<td>The maximum time in ms for a connection to stay idle before failing.</td>
</tr>
+ <tr>
+ <td><h5>rest.profiling.dir</h5></td>
+ <td style="word-wrap: break-word;">System.getProperty("java.io.tmpdir")</td>
+ <td>String</td>
+ <td>Profiling result storing directory.</td>
+ </tr>
+ <tr>
+ <td><h5>rest.profiling.duration-max</h5></td>
+ <td style="word-wrap: break-word;">5 min</td>
+ <td>Duration</td>
+ <td>Maximum profiling duration for each profiling request. Any profiling request's duration exceeding this value will not be accepted.</td>
+ </tr>
+ <tr>
+ <td><h5>rest.profiling.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Enables the experimental profiler feature.</td>
+ </tr>
+ <tr>
+ <td><h5>rest.profiling.history-size</h5></td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>Maximum profiling history instance to be maintained for JobManager or each TaskManager. The oldest instance will be removed on a rolling basis when the history size exceeds this value.</td>
+ </tr>
<tr>
<td><h5>rest.retry.delay</h5></td>
<td style="word-wrap: break-word;">3000</td>
diff --git a/docs/layouts/shortcodes/generated/rest_configuration.html b/docs/layouts/shortcodes/generated/rest_configuration.html
index af006be2037..0af3edc93f7 100644
--- a/docs/layouts/shortcodes/generated/rest_configuration.html
+++ b/docs/layouts/shortcodes/generated/rest_configuration.html
@@ -116,6 +116,30 @@
<td>Integer</td>
<td>The port that the client connects to. If rest.bind-port has not been specified, then the REST server will bind to this port. Attention: This option is respected only if the high-availability configuration is NONE.</td>
</tr>
+ <tr>
+ <td><h5>rest.profiling.dir</h5></td>
+ <td style="word-wrap: break-word;">System.getProperty("java.io.tmpdir")</td>
+ <td>String</td>
+ <td>Profiling result storing directory.</td>
+ </tr>
+ <tr>
+ <td><h5>rest.profiling.duration-max</h5></td>
+ <td style="word-wrap: break-word;">5 min</td>
+ <td>Duration</td>
+ <td>Maximum profiling duration for each profiling request. Any profiling request's duration exceeding this value will not be accepted.</td>
+ </tr>
+ <tr>
+ <td><h5>rest.profiling.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Enables the experimental profiler feature.</td>
+ </tr>
+ <tr>
+ <td><h5>rest.profiling.history-size</h5></td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>Maximum profiling history instance to be maintained for JobManager or each TaskManager. The oldest instance will be removed on a rolling basis when the history size exceeds this value.</td>
+ </tr>
<tr>
<td><h5>rest.retry.delay</h5></td>
<td style="word-wrap: break-word;">3000</td>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java
index a4db7011787..a9f9d6343cd 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java
@@ -295,4 +295,41 @@ public class RestOptions {
.defaultValue(Duration.ofMinutes(5))
.withDescription(
"Maximum duration that the result of an async operation is stored. Once elapsed the result of the operation can no longer be retrieved.");
+
+ /** Enables the experimental profiler feature. */
+ @Documentation.Section(Documentation.Sections.EXPERT_REST)
+ public static final ConfigOption<Boolean> ENABLE_PROFILER =
+ key("rest.profiling.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enables the experimental profiler feature.");
+
+ /** Maximum history size of profiling list. */
+ @Documentation.Section(Documentation.Sections.EXPERT_REST)
+ public static final ConfigOption<Integer> MAX_PROFILING_HISTORY_SIZE =
+ key("rest.profiling.history-size")
+ .intType()
+ .defaultValue(10)
+ .withDescription(
+ "Maximum profiling history instance to be maintained for JobManager or each TaskManager. "
+ + "The oldest instance will be removed on a rolling basis when the history size exceeds this value.");
+
+ /** Maximum profiling duration for profiling function. */
+ @Documentation.Section(Documentation.Sections.EXPERT_REST)
+ public static final ConfigOption<Duration> MAX_PROFILING_DURATION =
+ key("rest.profiling.duration-max")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(300))
+ .withDescription(
+ "Maximum profiling duration for each profiling request. "
+ + "Any profiling request's duration exceeding this value will not be accepted.");
+
+ /** Directory for storing the profiling results. */
+ @Documentation.Section(Documentation.Sections.EXPERT_REST)
+ @Documentation.OverrideDefault("System.getProperty(\"java.io.tmpdir\")")
+ public static final ConfigOption<String> PROFILING_RESULT_DIR =
+ key("rest.profiling.dir")
+ .stringType()
+ .defaultValue(System.getProperty("java.io.tmpdir"))
+ .withDescription("Profiling result storing directory.");
}
diff --git a/flink-dist/src/main/resources/META-INF/NOTICE b/flink-dist/src/main/resources/META-INF/NOTICE
index a5358c47c30..e1554ae5a8a 100644
--- a/flink-dist/src/main/resources/META-INF/NOTICE
+++ b/flink-dist/src/main/resources/META-INF/NOTICE
@@ -20,6 +20,7 @@ This project bundles the following dependencies under the Apache Software Licens
- org.lz4:lz4-java:1.8.0
- org.objenesis:objenesis:2.1
- org.xerial.snappy:snappy-java:1.1.10.4
+- tools.profiler:async-profiler:2.9
This project bundles the following dependencies under the BSD license.
See bundled license files for details.
diff --git a/flink-runtime-web/web-dashboard/src/app/components/humanize-watermark.pipe.ts b/flink-runtime-web/web-dashboard/src/app/components/humanize-watermark.pipe.ts
index 140cac44e0c..938cfb8cb16 100644
--- a/flink-runtime-web/web-dashboard/src/app/components/humanize-watermark.pipe.ts
+++ b/flink-runtime-web/web-dashboard/src/app/components/humanize-watermark.pipe.ts
@@ -44,8 +44,8 @@ export class HumanizeWatermarkToDatetimePipe implements PipeTransform {
constructor(private readonly configService: ConfigService) {}
public transform(value: number): number | string {
- if (isNaN(value) || value <= this.configService.LONG_MIN_VALUE) {
- return '-';
+ if (value == null || isNaN(value) || value <= this.configService.LONG_MIN_VALUE) {
+ return 'N/A';
} else {
return new Date(value).toLocaleString();
}
diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/job-profiler.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/job-profiler.ts
new file mode 100644
index 00000000000..e584b3cb4be
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/interfaces/job-profiler.ts
@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+
+export interface ProfilingList {
+ profilingList: ProfilingDetail[];
+}
+
+export interface ProfilingDetail {
+ status: string;
+ triggerTime: number;
+ finishedTime: number;
+ duration: number;
+ message: string;
+ outputFile: string;
+ mode: string;
+}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/job-manager.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/job-manager.component.ts
index 2180ffc2ab4..036ec91206a 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/job-manager.component.ts
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/job-manager.component.ts
@@ -36,6 +36,7 @@ export class JobManagerComponent {
{ path: 'logs', title: 'Logs' },
{ path: 'stdout', title: 'Stdout' },
{ path: 'log', title: 'Log List' },
- { path: 'thread-dump', title: 'Thread Dump' }
+ { path: 'thread-dump', title: 'Thread Dump' },
+ { path: 'profiler', title: 'Profiler' }
];
}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.html
new file mode 100644
index 00000000000..ab46841f9c6
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.html
@@ -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.
+ -->
+
+<nz-card nzSize="small" style="width: 100%; height: 100%">
+ <div style="margin-bottom: 10px">
+ <form nz-form [nzLayout]="'inline'">
+ <nz-form-item>
+ <nz-form-label nzFor="duration">Profiling Duration</nz-form-label>
+ <nz-form-control>
+ <nz-input-number
+ [(ngModel)]="duration"
+ [nzMin]="1"
+ [nzStep]="30"
+ [nzFormatter]="formatterDuration"
+ [nzParser]="parserDuration"
+ nzPlaceHolder="Duration"
+ name="duration"
+ ></nz-input-number>
+ </nz-form-control>
+ </nz-form-item>
+ <nz-form-item>
+ <nz-form-control>
+ <button
+ nz-button
+ nzType="primary"
+ [nzLoading]="isCreating"
+ [disabled]="!isEnabled"
+ (click)="createProfilingInstance()"
+ style="margin-left: 10px"
+ >
+ Create Profiling Instance
+ </button>
+ <i
+ class="header-icon"
+ style="margin-left: 10px"
+ nz-icon
+ nz-tooltip
+ [nzTooltipTitle]="titleTemplate"
+ nzTooltipTitle=""
+ nzType="info-circle"
+ ></i>
+ <ng-template #titleTemplate>
+ <span>
+ Please refer to
+ <a href="https://github.com/async-profiler/async-profiler/wiki">
+ async-profiler's wiki
+ </a>
+ for more detailed info of this feature.
+ </span>
+ </ng-template>
+ </nz-form-control>
+ </nz-form-item>
+ </form>
+ <nz-alert
+ *ngIf="!isEnabled"
+ nzType="warning"
+ style="margin-top: 10px"
+ nzShowIcon
+ nzMessage="Please set the config `rest.profiling.enabled=true` to enable this experimental profiler feature."
+ ></nz-alert>
+ </div>
+ <nz-table
+ [nzSize]="'small'"
+ [nzData]="profilingList"
+ [nzLoading]="isLoading"
+ [nzFrontPagination]="false"
+ [nzShowPagination]="false"
+ >
+ <thead>
+ <tr>
+ <th nzWidth="5%">Index</th>
+ <th nzWidth="15%">Trigger Time</th>
+ <th nzWidth="15%">Finished Time</th>
+ <th nzWidth="10%">Profiling Duration</th>
+ <th nzWidth="5%">Mode</th>
+ <th nzWidth="10%">Status</th>
+ <th nzWidth="25%">Link</th>
+ <th nzWidth="15%">Message</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr *ngFor="let info of profilingList; let id = index">
+ <td>{{ id }}</td>
+ <td>{{ info.triggerTime | humanizeWatermarkToDatetime }}</td>
+ <td>{{ info.finishedTime | humanizeWatermarkToDatetime }}</td>
+ <td>{{ info.duration }} s</td>
+ <td>{{ info.mode }}</td>
+ <td>{{ info.status }}</td>
+ <td>
+ <a (click)="downloadProfilingResult(info.outputFile)">{{ info.outputFile }}</a>
+ </td>
+ <td>{{ info.message }}</td>
+ </tr>
+ </tbody>
+ </nz-table>
+</nz-card>
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.less b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.less
new file mode 100644
index 00000000000..3b5de2d4f7f
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.less
@@ -0,0 +1,43 @@
+/*
+ * 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 "theme";
+
+:host {
+ display: flex;
+ flex: 1;
+
+ ::ng-deep {
+ .ant-table-cell {
+ font-size: @font-size-sm;
+ }
+
+ ::-webkit-scrollbar {
+ display: none;
+ }
+
+ nz-table,
+ nz-spin,
+ cdk-virtual-scroll-viewport,
+ nz-table-inner-scroll,
+ .ant-spin-container,
+ .ant-table {
+ height: 100%;
+ }
+ }
+}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.ts
new file mode 100644
index 00000000000..c17e88bba5b
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.ts
@@ -0,0 +1,151 @@
+/*
+ * 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 { CommonModule } from '@angular/common';
+import { ChangeDetectionStrategy, ChangeDetectorRef, Component, OnDestroy, OnInit } from '@angular/core';
+import { FormsModule } from '@angular/forms';
+import { Subject } from 'rxjs';
+import { mergeMap, startWith, takeUntil } from 'rxjs/operators';
+
+import {
+ HumanizeWatermarkPipe,
+ HumanizeWatermarkToDatetimePipe
+} from '@flink-runtime-web/components/humanize-watermark.pipe';
+import { ProfilingDetail } from '@flink-runtime-web/interfaces/job-profiler';
+import { JobManagerService, StatusService } from '@flink-runtime-web/services';
+import { NzAlertModule } from 'ng-zorro-antd/alert';
+import { NzButtonModule } from 'ng-zorro-antd/button';
+import { NzCardModule } from 'ng-zorro-antd/card';
+import { NzFormModule } from 'ng-zorro-antd/form';
+import { NzIconModule } from 'ng-zorro-antd/icon';
+import { NzInputNumberModule } from 'ng-zorro-antd/input-number';
+import { NzMessageModule, NzMessageService } from 'ng-zorro-antd/message';
+import { NzSelectModule } from 'ng-zorro-antd/select';
+import { NzSpaceModule } from 'ng-zorro-antd/space';
+import { NzTableModule } from 'ng-zorro-antd/table';
+import { NzToolTipModule } from 'ng-zorro-antd/tooltip';
+
+@Component({
+ selector: 'flink-job-manager-profiler',
+ templateUrl: './job-manager-profiler.component.html',
+ styleUrls: ['./job-manager-profiler.component.less'],
+ changeDetection: ChangeDetectionStrategy.OnPush,
+ imports: [
+ NzCardModule,
+ NzFormModule,
+ NzInputNumberModule,
+ HumanizeWatermarkPipe,
+ FormsModule,
+ NzButtonModule,
+ NzAlertModule,
+ NzTableModule,
+ NzMessageModule,
+ CommonModule,
+ NzSpaceModule,
+ HumanizeWatermarkToDatetimePipe,
+ NzSelectModule,
+ NzToolTipModule,
+ NzIconModule
+ ],
+ standalone: true
+})
+export class JobManagerProfilerComponent implements OnInit, OnDestroy {
+ private readonly destroy$ = new Subject<void>();
+ profilingList: ProfilingDetail[] = [];
+ isLoading = true;
+ isCreating = false;
+ duration = 30;
+ selectMode = 'CPU';
+ isEnabled = false;
+ formatterDuration = (value: number): string => `${value} s`;
+ parserDuration = (value: string): string => value.replace(' s', '');
+
+ constructor(
+ private jobManagerService: JobManagerService,
+ private readonly statusService: StatusService,
+ private message: NzMessageService,
+ private cdr: ChangeDetectorRef
+ ) {}
+
+ public createProfilingInstance(): void {
+ if (this.profilingList.length > 0 && this.profilingList[0].status === 'RUNNING') {
+ this.message.warning('Please wait for last profiling finished.');
+ return;
+ }
+ this.isCreating = true;
+ this.jobManagerService.createProfilingInstance(this.selectMode, this.duration).subscribe({
+ next: profilingDetail => {
+ this.profilingList.unshift(profilingDetail);
+ this.isCreating = false;
+ this.cdr.markForCheck();
+ },
+ error: () => {
+ this.isCreating = false;
+ this.cdr.markForCheck();
+ }
+ });
+ }
+
+ public ngOnInit(): void {
+ this.statusService.refresh$
+ .pipe(
+ startWith(true),
+ mergeMap(() => {
+ this.isLoading = true;
+ this.cdr.markForCheck();
+ return this.jobManagerService.loadProfilingList();
+ }),
+ takeUntil(this.destroy$)
+ )
+ .subscribe({
+ next: data => {
+ this.profilingList = data.profilingList;
+ this.isLoading = false;
+ this.isEnabled = true;
+ this.cdr.markForCheck();
+ },
+ error: () => {
+ this.isLoading = false;
+ this.destroy$.next();
+ this.destroy$.complete();
+ this.cdr.markForCheck();
+ }
+ });
+ }
+
+ public downloadProfilingResult(filePath: string): void {
+ this.isLoading = true;
+ this.cdr.markForCheck();
+ this.jobManagerService.loadProfilingResult(filePath).subscribe({
+ next: data => {
+ const anchor = document.createElement('a');
+ anchor.href = data.url;
+ anchor.download = data.url;
+ document.body.appendChild(anchor);
+ anchor.click();
+ },
+ complete: () => {
+ this.isLoading = false;
+ this.cdr.markForCheck();
+ }
+ });
+ }
+
+ public ngOnDestroy(): void {
+ this.destroy$.next();
+ this.destroy$.complete();
+ }
+}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/routes.ts b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/routes.ts
index b2ee7f8e689..7cd77f12638 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/routes.ts
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/routes.ts
@@ -77,6 +77,14 @@ export const JOB_MANAGER_ROUTES: Routes = [
path: 'thread-dump'
}
},
+ {
+ path: 'profiler',
+ loadComponent: () =>
+ import('./profiler/job-manager-profiler.component').then(m => m.JobManagerProfilerComponent),
+ data: {
+ path: 'profiler'
+ }
+ },
{
path: '**',
redirectTo: 'metrics',
diff --git a/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts b/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts
index 32d809304d0..28eb337222f 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.ts
@@ -30,6 +30,7 @@ import {
ClusterConfiguration,
EnvironmentInfo
} from '@flink-runtime-web/interfaces';
+import { ProfilingDetail, ProfilingList } from '@flink-runtime-web/interfaces/job-profiler';
import { ConfigService } from './config.service';
@@ -125,4 +126,27 @@ export class JobManagerService {
.get<{ url: string }>(`${this.configService.BASE_URL}/jobs/${jobId}/jobmanager/log-url`)
.pipe(map(data => data.url));
}
+
+ loadProfilingList(): Observable<ProfilingList> {
+ return this.httpClient.get<ProfilingList>(`${this.configService.BASE_URL}/jobmanager/profiler`);
+ }
+
+ createProfilingInstance(mode: string, duration: number): Observable<ProfilingDetail> {
+ const requestParam = { mode, duration };
+ return this.httpClient.post<ProfilingDetail>(`${this.configService.BASE_URL}/jobmanager/profiler`, requestParam);
+ }
+
+ loadProfilingResult(filePath: string): Observable<Record<string, string>> {
+ const url = `${this.configService.BASE_URL}/jobmanager/profiler/${filePath}`;
+ return this.httpClient
+ .get(url, { responseType: 'text', headers: new HttpHeaders().append('Cache-Control', 'no-cache') })
+ .pipe(
+ map(data => {
+ return {
+ data,
+ url
+ };
+ })
+ );
+ }
}
diff --git a/flink-runtime/pom.xml b/flink-runtime/pom.xml
index c6bd2e90964..a4fd54ac5e5 100644
--- a/flink-runtime/pom.xml
+++ b/flink-runtime/pom.xml
@@ -215,6 +215,13 @@ under the License.
<artifactId>snappy-java</artifactId>
</dependency>
+ <!-- Async Profiler library -->
+ <dependency>
+ <groupId>tools.profiler</groupId>
+ <artifactId>async-profiler</artifactId>
+ <version>2.9</version>
+ </dependency>
+
<!-- Lz4 compression library -->
<dependency>
<groupId>org.lz4</groupId>
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingFileHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingFileHandler.java
new file mode 100644
index 00000000000..04c6831ce6f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingFileHandler.java
@@ -0,0 +1,63 @@
+/*
+ * 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.flink.runtime.rest.handler.cluster;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.ProfilingFileNamePathParameter;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
+import org.apache.flink.runtime.rest.messages.cluster.ProfilingFileMessageParameters;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+
+import java.io.File;
+import java.util.Map;
+
+/** Rest handler which serves the profiler result file from JobManager. */
+public class JobManagerProfilingFileHandler
+ extends AbstractJobManagerFileHandler<ProfilingFileMessageParameters> {
+
+ private final String profilingResultDir;
+
+ public JobManagerProfilingFileHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ UntypedResponseMessageHeaders<EmptyRequestBody, ProfilingFileMessageParameters>
+ messageHeaders,
+ Configuration configs) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+
+ this.profilingResultDir = configs.getString(RestOptions.PROFILING_RESULT_DIR);
+ }
+
+ @Override
+ protected File getFile(HandlerRequest<EmptyRequestBody> handlerRequest) {
+ if (profilingResultDir == null) {
+ return null;
+ }
+ String filename =
+ new File(handlerRequest.getPathParameter(ProfilingFileNamePathParameter.class))
+ .getName();
+ return new File(profilingResultDir, filename);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingHandler.java
new file mode 100644
index 00000000000..7072c9fd129
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingHandler.java
@@ -0,0 +1,77 @@
+/*
+ * 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.flink.runtime.rest.handler.cluster;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.ProfilingInfo;
+import org.apache.flink.runtime.rest.messages.cluster.ProfilingRequestBody;
+import org.apache.flink.runtime.util.profiler.ProfilingService;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Rest handler which serves the profiling info from the JobManager. */
+public class JobManagerProfilingHandler
+ extends AbstractRestHandler<
+ RestfulGateway, ProfilingRequestBody, ProfilingInfo, EmptyMessageParameters> {
+ private final long maxDurationInSeconds;
+ private final ProfilingService profilingService;
+
+ public JobManagerProfilingHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ MessageHeaders<ProfilingRequestBody, ProfilingInfo, EmptyMessageParameters>
+ messageHeaders,
+ final Configuration configuration) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+ this.maxDurationInSeconds =
+ configuration.get(RestOptions.MAX_PROFILING_DURATION).getSeconds();
+ this.profilingService = ProfilingService.getInstance(configuration);
+ }
+
+ @Override
+ protected CompletableFuture<ProfilingInfo> handleRequest(
+ @Nonnull HandlerRequest<ProfilingRequestBody> request, @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ ProfilingRequestBody profilingRequest = request.getRequestBody();
+ int duration = profilingRequest.getDuration();
+ if (duration <= 0 || duration > maxDurationInSeconds) {
+ return FutureUtils.completedExceptionally(
+ new IllegalArgumentException(
+ String.format(
+ "`duration` must be set between (0s, %ds].",
+ maxDurationInSeconds)));
+ }
+ return profilingService.requestProfiling(
+ "JobManager", duration, profilingRequest.getMode());
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingListHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingListHandler.java
new file mode 100644
index 00000000000..16bca0c93d8
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerProfilingListHandler.java
@@ -0,0 +1,63 @@
+/*
+ * 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.flink.runtime.rest.handler.cluster;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.ProfilingInfoList;
+import org.apache.flink.runtime.util.profiler.ProfilingService;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Rest handler which serves the profiling list from the JobManager. */
+public class JobManagerProfilingListHandler
+ extends AbstractRestHandler<
+ RestfulGateway, EmptyRequestBody, ProfilingInfoList, EmptyMessageParameters> {
+
+ private final ProfilingService profilingService;
+
+ public JobManagerProfilingListHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ MessageHeaders<EmptyRequestBody, ProfilingInfoList, EmptyMessageParameters>
+ messageHeaders,
+ final Configuration configuration) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+ this.profilingService = ProfilingService.getInstance(configuration);
+ }
+
+ @Override
+ protected CompletableFuture<ProfilingInfoList> handleRequest(
+ @Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ return profilingService.getProfilingList("JobManager").thenApply(ProfilingInfoList::new);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingFileNamePathParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingFileNamePathParameter.java
new file mode 100644
index 00000000000..fcdab49ee96
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingFileNamePathParameter.java
@@ -0,0 +1,46 @@
+/*
+ * 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.flink.runtime.rest.messages;
+
+import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingFileHandler;
+
+/** Parameters used by {@link JobManagerProfilingFileHandler}. */
+public class ProfilingFileNamePathParameter extends MessagePathParameter<String> {
+
+ public static final String KEY = "filename";
+
+ public ProfilingFileNamePathParameter() {
+ super(KEY);
+ }
+
+ @Override
+ protected String convertFromString(String value) {
+ return value;
+ }
+
+ @Override
+ protected String convertToString(String value) {
+ return value;
+ }
+
+ @Override
+ public String getDescription() {
+ return "String value that identifies the file name from which to read.";
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingInfo.java
new file mode 100644
index 00000000000..0c8c1f8003f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingInfo.java
@@ -0,0 +1,187 @@
+/*
+ * 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.flink.runtime.rest.messages;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+
+/** Contains information of a Profiling Instance. */
+public class ProfilingInfo implements ResponseBody, Serializable {
+ private static final long serialVersionUID = 1L;
+ public static final String FIELD_NAME_STATUS = "status";
+ public static final String FIELD_NAME_MODE = "mode";
+ public static final String FIELD_NAME_TRIGGER_TIME = "triggerTime";
+ public static final String FIELD_NAME_FINISHED_TIME = "finishedTime";
+ public static final String FIELD_NAME_DURATION = "duration";
+ public static final String FIELD_NAME_MESSAGE = "message";
+ public static final String FIELD_NAME_OUTPUT_FILE = "outputFile";
+
+ @JsonProperty(FIELD_NAME_STATUS)
+ private ProfilingStatus status;
+
+ @JsonProperty(FIELD_NAME_MODE)
+ private ProfilingMode mode;
+
+ @JsonProperty(FIELD_NAME_TRIGGER_TIME)
+ private Long triggerTime;
+
+ @JsonProperty(FIELD_NAME_FINISHED_TIME)
+ private Long finishedTime;
+
+ @JsonProperty(FIELD_NAME_DURATION)
+ private Long duration;
+
+ @JsonProperty(FIELD_NAME_MESSAGE)
+ private String message;
+
+ @JsonProperty(FIELD_NAME_OUTPUT_FILE)
+ private String outputFile;
+
+ private ProfilingInfo() {}
+
+ private ProfilingInfo(
+ ProfilingStatus status,
+ ProfilingMode mode,
+ long triggerTime,
+ long finishedTime,
+ long duration,
+ String message,
+ String outputFile) {
+ this.status = status;
+ this.mode = mode;
+ this.triggerTime = triggerTime;
+ this.finishedTime = finishedTime;
+ this.duration = duration;
+ this.message = message;
+ this.outputFile = outputFile;
+ }
+
+ public ProfilingInfo fail(String message) {
+ this.status = ProfilingStatus.FAILED;
+ this.message = message;
+ this.finishedTime = System.currentTimeMillis();
+ return this;
+ }
+
+ public ProfilingInfo success(String outputFile) {
+ this.status = ProfilingStatus.FINISHED;
+ this.finishedTime = System.currentTimeMillis();
+ this.outputFile = outputFile;
+ this.message = "Profiling Successful";
+ return this;
+ }
+
+ public ProfilingStatus getStatus() {
+ return status;
+ }
+
+ public ProfilingMode getProfilingMode() {
+ return mode;
+ }
+
+ public Long getDuration() {
+ return duration;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+ public Long getFinishedTime() {
+ return finishedTime;
+ }
+
+ public String getOutputFile() {
+ return outputFile;
+ }
+
+ public long getTriggerTime() {
+ return triggerTime;
+ }
+
+ public static ProfilingInfo create(long duration, ProfilingMode mode) {
+ ProfilingInfo profilingInfo = new ProfilingInfo();
+ profilingInfo.mode = mode;
+ profilingInfo.triggerTime = System.currentTimeMillis();
+ profilingInfo.status = ProfilingStatus.RUNNING;
+ profilingInfo.duration = duration;
+ return profilingInfo;
+ }
+
+ @JsonCreator
+ public static ProfilingInfo create(
+ @JsonProperty(FIELD_NAME_STATUS) ProfilingStatus status,
+ @JsonProperty(FIELD_NAME_MODE) ProfilingMode mode,
+ @JsonProperty(FIELD_NAME_TRIGGER_TIME) long triggerTime,
+ @JsonProperty(FIELD_NAME_FINISHED_TIME) long finishedTime,
+ @JsonProperty(FIELD_NAME_DURATION) long duration,
+ @JsonProperty(FIELD_NAME_MESSAGE) String message,
+ @JsonProperty(FIELD_NAME_OUTPUT_FILE) String outputPath) {
+ return new ProfilingInfo(
+ status, mode, triggerTime, finishedTime, duration, message, outputPath);
+ }
+
+ @Override
+ public String toString() {
+ return "ProfilingInfo{"
+ + "status="
+ + status
+ + ", mode="
+ + mode
+ + ", triggerTime="
+ + triggerTime
+ + ", finishedTime="
+ + finishedTime
+ + ", duration="
+ + duration
+ + ", message='"
+ + message
+ + '\''
+ + ", outputFile='"
+ + outputFile
+ + '\''
+ + '}';
+ }
+
+ /** Profiling Status. */
+ public enum ProfilingStatus {
+ RUNNING,
+ FINISHED,
+ FAILED;
+
+ public boolean isRunning() {
+ return this == RUNNING;
+ }
+ }
+
+ /** Supported profiling mode in async-profiler. */
+ public enum ProfilingMode {
+ CPU,
+ ALLOC,
+ LOCK,
+ WALL,
+ ITIMER;
+
+ public String getCode() {
+ return this.name().toLowerCase();
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingInfoList.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingInfoList.java
new file mode 100644
index 00000000000..6cac04887f9
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ProfilingInfoList.java
@@ -0,0 +1,79 @@
+/*
+ * 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.flink.runtime.rest.messages;
+
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Objects;
+
+/** Class containing a collection of {@link ProfilingInfo}. */
+public class ProfilingInfoList implements ResponseBody, Serializable {
+
+ public static final String FIELD_NAME_FLAME_GRAPHS = "profilingList";
+
+ private static final long serialVersionUID = 1L;
+
+ @JsonProperty(FIELD_NAME_FLAME_GRAPHS)
+ private final Collection<ProfilingInfo> profilingInfos;
+
+ @JsonCreator
+ public ProfilingInfoList(
+ @JsonProperty(FIELD_NAME_FLAME_GRAPHS) Collection<ProfilingInfo> profilingInfos) {
+ this.profilingInfos = Preconditions.checkNotNull(profilingInfos);
+ }
+
+ public Collection<ProfilingInfo> getProfilingInfos() {
+ return profilingInfos;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ProfilingInfoList that = (ProfilingInfoList) o;
+ return Objects.equals(profilingInfos, that.profilingInfos);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(profilingInfos);
+ }
+
+ public static ProfilingInfoList empty() {
+ return new ProfilingInfoList(Collections.emptyList());
+ }
+
+ public static ProfilingInfoList createNullable(Collection<ProfilingInfo> profilingInfos) {
+ if (CollectionUtil.isNullOrEmpty(profilingInfos)) {
+ return empty();
+ }
+ return new ProfilingInfoList(profilingInfos);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingFileHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingFileHeaders.java
new file mode 100644
index 00000000000..576a09251c1
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingFileHeaders.java
@@ -0,0 +1,63 @@
+/*
+ * 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.flink.runtime.rest.messages.cluster;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingFileHandler;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.ProfilingFileNamePathParameter;
+import org.apache.flink.runtime.rest.messages.RuntimeUntypedResponseMessageHeaders;
+
+/** Headers for the {@link JobManagerProfilingFileHandler}. */
+public class JobManagerProfilingFileHeaders
+ implements RuntimeUntypedResponseMessageHeaders<
+ EmptyRequestBody, ProfilingFileMessageParameters> {
+
+ private static final JobManagerProfilingFileHeaders INSTANCE =
+ new JobManagerProfilingFileHeaders();
+
+ private static final String URL =
+ String.format("/jobmanager/profiler/:%s", ProfilingFileNamePathParameter.KEY);
+
+ private JobManagerProfilingFileHeaders() {}
+
+ public static JobManagerProfilingFileHeaders getInstance() {
+ return INSTANCE;
+ }
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public ProfilingFileMessageParameters getUnresolvedMessageParameters() {
+ return new ProfilingFileMessageParameters();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.GET;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return URL;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingHeaders.java
new file mode 100644
index 00000000000..6b241eb9bf0
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingHeaders.java
@@ -0,0 +1,87 @@
+/*
+ * 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.flink.runtime.rest.messages.cluster;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingHandler;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.ProfilingInfo;
+import org.apache.flink.runtime.rest.versioning.RestAPIVersion;
+import org.apache.flink.runtime.rest.versioning.RuntimeRestAPIVersion;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/** Headers for the {@link JobManagerProfilingHandler}. */
+public class JobManagerProfilingHeaders
+ implements MessageHeaders<ProfilingRequestBody, ProfilingInfo, EmptyMessageParameters> {
+
+ private static final JobManagerProfilingHeaders INSTANCE = new JobManagerProfilingHeaders();
+
+ private static final String URL = "/jobmanager/profiler";
+
+ private JobManagerProfilingHeaders() {}
+
+ public static JobManagerProfilingHeaders getInstance() {
+ return INSTANCE;
+ }
+
+ @Override
+ public Class<ProfilingInfo> getResponseClass() {
+ return ProfilingInfo.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Create a profiling instance on JobManager.";
+ }
+
+ @Override
+ public Class<ProfilingRequestBody> getRequestClass() {
+ return ProfilingRequestBody.class;
+ }
+
+ @Override
+ public EmptyMessageParameters getUnresolvedMessageParameters() {
+ return EmptyMessageParameters.getInstance();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.POST;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return URL;
+ }
+
+ @Override
+ public Collection<? extends RestAPIVersion<?>> getSupportedAPIVersions() {
+ return Collections.singleton(RuntimeRestAPIVersion.V1);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingListHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingListHeaders.java
new file mode 100644
index 00000000000..fdf43856159
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/JobManagerProfilingListHeaders.java
@@ -0,0 +1,89 @@
+/*
+ * 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.flink.runtime.rest.messages.cluster;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingHandler;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.ProfilingInfoList;
+import org.apache.flink.runtime.rest.versioning.RestAPIVersion;
+import org.apache.flink.runtime.rest.versioning.RuntimeRestAPIVersion;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/** Headers for the {@link JobManagerProfilingHandler}. */
+public class JobManagerProfilingListHeaders
+ implements MessageHeaders<EmptyRequestBody, ProfilingInfoList, EmptyMessageParameters> {
+
+ private static final JobManagerProfilingListHeaders INSTANCE =
+ new JobManagerProfilingListHeaders();
+
+ private static final String URL = "/jobmanager/profiler";
+
+ private JobManagerProfilingListHeaders() {}
+
+ public static JobManagerProfilingListHeaders getInstance() {
+ return INSTANCE;
+ }
+
+ @Override
+ public Class<ProfilingInfoList> getResponseClass() {
+ return ProfilingInfoList.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Returns the profiling history on JobManager.";
+ }
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public EmptyMessageParameters getUnresolvedMessageParameters() {
+ return EmptyMessageParameters.getInstance();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.GET;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return URL;
+ }
+
+ @Override
+ public Collection<? extends RestAPIVersion<?>> getSupportedAPIVersions() {
+ return Collections.singleton(RuntimeRestAPIVersion.V1);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ProfilingFileMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ProfilingFileMessageParameters.java
new file mode 100644
index 00000000000..f08590048c4
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ProfilingFileMessageParameters.java
@@ -0,0 +1,45 @@
+/*
+ * 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.flink.runtime.rest.messages.cluster;
+
+import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingFileHandler;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.MessagePathParameter;
+import org.apache.flink.runtime.rest.messages.MessageQueryParameter;
+import org.apache.flink.runtime.rest.messages.ProfilingFileNamePathParameter;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/** Parameters for {@link JobManagerProfilingFileHandler}. */
+public class ProfilingFileMessageParameters extends MessageParameters {
+
+ public final ProfilingFileNamePathParameter profilingFileNamePathParameter =
+ new ProfilingFileNamePathParameter();
+
+ @Override
+ public Collection<MessagePathParameter<?>> getPathParameters() {
+ return Collections.singleton(profilingFileNamePathParameter);
+ }
+
+ @Override
+ public Collection<MessageQueryParameter<?>> getQueryParameters() {
+ return Collections.emptyList();
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ProfilingRequestBody.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ProfilingRequestBody.java
new file mode 100644
index 00000000000..862bc27fba8
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/cluster/ProfilingRequestBody.java
@@ -0,0 +1,57 @@
+/*
+ * 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.flink.runtime.rest.messages.cluster;
+
+import org.apache.flink.runtime.rest.messages.ProfilingInfo;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** {@link RequestBody} to trigger profiling instance. */
+public class ProfilingRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_MODE = "mode";
+ private static final String FIELD_NAME_DURATION = "duration";
+
+ @JsonProperty(FIELD_NAME_MODE)
+ private final ProfilingInfo.ProfilingMode mode;
+
+ @JsonProperty(FIELD_NAME_DURATION)
+ private final int duration;
+
+ @JsonCreator
+ public ProfilingRequestBody(
+ @JsonProperty(FIELD_NAME_MODE) final ProfilingInfo.ProfilingMode mode,
+ @JsonProperty(FIELD_NAME_DURATION) final int duration) {
+ this.mode = mode;
+ this.duration = duration;
+ }
+
+ @JsonIgnore
+ public ProfilingInfo.ProfilingMode getMode() {
+ return mode;
+ }
+
+ @JsonIgnore
+ public int getDuration() {
+ return duration;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/profiler/ProfilingService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/profiler/ProfilingService.java
new file mode 100644
index 00000000000..96cbae0a290
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/profiler/ProfilingService.java
@@ -0,0 +1,237 @@
+/*
+ * 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.flink.runtime.util.profiler;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.runtime.rest.messages.ProfilingInfo;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import one.profiler.AsyncProfiler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.text.SimpleDateFormat;
+import java.util.ArrayDeque;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/** Create and keep profiling requests with rolling policy. */
+public class ProfilingService implements Closeable {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(ProfilingService.class);
+ private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd_HH_mm_ss");
+ private static volatile ProfilingService instance;
+ private final Map<String, ArrayDeque<ProfilingInfo>> profilingMap;
+ private final String profilingResultDir;
+ private final int historySizeLimit;
+ private final ScheduledExecutorService scheduledExecutor;
+ private ProfilingFuture profilingFuture;
+
+ private ProfilingService(Configuration configs) {
+ this.profilingMap = new HashMap<>();
+ this.historySizeLimit = configs.getInteger(RestOptions.MAX_PROFILING_HISTORY_SIZE);
+ Preconditions.checkArgument(
+ historySizeLimit > 0,
+ String.format(
+ "Configured %s must be positive.",
+ RestOptions.MAX_PROFILING_HISTORY_SIZE.key()));
+ this.profilingResultDir = configs.getString(RestOptions.PROFILING_RESULT_DIR);
+ this.scheduledExecutor =
+ Executors.newSingleThreadScheduledExecutor(
+ new ExecutorThreadFactory.Builder()
+ .setPoolName("flink-profiling-service")
+ .build());
+ }
+
+ public static ProfilingService getInstance(Configuration configs) {
+ if (instance == null) {
+ synchronized (ProfilingService.class) {
+ if (instance == null) {
+ instance = new ProfilingService(configs);
+ }
+ }
+ }
+ return instance;
+ }
+
+ public CompletableFuture<ProfilingInfo> requestProfiling(
+ String resourceID, long duration, ProfilingInfo.ProfilingMode mode) {
+ if (profilingFuture != null && !profilingFuture.isDone()) {
+ return FutureUtils.completedExceptionally(
+ new IllegalStateException(resourceID + " is still under profiling."));
+ }
+ ProfilingInfo profilingInfo = ProfilingInfo.create(duration, mode);
+ profilingMap.putIfAbsent(resourceID, new ArrayDeque<>());
+ profilingMap.get(resourceID).addFirst(profilingInfo);
+ AsyncProfiler profiler = AsyncProfiler.getInstance();
+ try {
+ String response =
+ profiler.execute(
+ ProfilerConstants.COMMAND_START.msg
+ + profilingInfo.getProfilingMode().getCode());
+ if (StringUtils.isNullOrWhitespaceOnly(response)
+ || !response.startsWith(ProfilerConstants.PROFILER_STARTED_SUCCESS.msg)) {
+ return CompletableFuture.completedFuture(
+ profilingInfo.fail("Start profiler failed. " + response));
+ }
+ } catch (Exception e) {
+ return CompletableFuture.completedFuture(
+ profilingInfo.fail("Start profiler failed. " + e));
+ }
+
+ scheduledExecutor.schedule(() -> stopProfiling(resourceID), duration, TimeUnit.SECONDS);
+
+ this.profilingFuture = new ProfilingFuture(duration, () -> stopProfiling(resourceID));
+ return CompletableFuture.completedFuture(profilingInfo);
+ }
+
+ private void stopProfiling(String resourceID) {
+ AsyncProfiler profiler = AsyncProfiler.getInstance();
+ ArrayDeque<ProfilingInfo> profilingList = profilingMap.get(resourceID);
+ Preconditions.checkState(!CollectionUtil.isNullOrEmpty(profilingList));
+ ProfilingInfo info = profilingList.getFirst();
+ try {
+ String fileName = formatOutputFileName(resourceID, info);
+ String outputPath = new File(profilingResultDir, fileName).getPath();
+ String response = profiler.execute(ProfilerConstants.COMMAND_STOP.msg + outputPath);
+ if (!StringUtils.isNullOrWhitespaceOnly(response)
+ && response.startsWith(ProfilerConstants.PROFILER_STOPPED_SUCCESS.msg)) {
+ info.success(fileName);
+ } else {
+ info.fail("Stop profiler failed. " + response);
+ }
+ rollingClearing(profilingList);
+ } catch (Throwable e) {
+ info.fail("Stop profiler failed. " + e);
+ }
+ }
+
+ private void rollingClearing(ArrayDeque<ProfilingInfo> profilingList) {
+ while (profilingList.size() > historySizeLimit) {
+ ProfilingInfo info = profilingList.pollLast();
+ String outputFile = info != null ? info.getOutputFile() : "";
+ if (StringUtils.isNullOrWhitespaceOnly(outputFile)) {
+ continue;
+ }
+ try {
+ Files.deleteIfExists(Paths.get(profilingResultDir, outputFile));
+ } catch (Exception e) {
+ LOG.error(String.format("Clearing file for %s failed. Skipped.", info), e);
+ }
+ }
+ }
+
+ private String formatOutputFileName(String resourceID, ProfilingInfo info) {
+ return String.format(
+ "%s_%s_%s.html", resourceID, info.getProfilingMode(), sdf.format(new Date()));
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ if (profilingFuture != null && !profilingFuture.isDone()) {
+ profilingFuture.cancel();
+ }
+ if (!scheduledExecutor.isShutdown()) {
+ scheduledExecutor.shutdownNow();
+ }
+ instance = null;
+ } catch (Exception e) {
+ LOG.error("Exception thrown during stopping profiling service. ", e);
+ }
+ }
+
+ public CompletableFuture<Collection<ProfilingInfo>> getProfilingList(String resourceID) {
+ return CompletableFuture.completedFuture(
+ profilingMap.getOrDefault(resourceID, new ArrayDeque<>()));
+ }
+
+ enum ProfilerConstants {
+ PROFILER_STARTED_SUCCESS("Profiling started"),
+ PROFILER_STOPPED_SUCCESS("OK"),
+ COMMAND_START("start,event="),
+ COMMAND_STOP("stop,file=");
+
+ private final String msg;
+
+ ProfilerConstants(String msg) {
+ this.msg = msg;
+ }
+ }
+
+ /**
+ * ProfilingFuture takes a handler and trigger duration, which will call the handler when
+ * timeout or cancelled.
+ */
+ class ProfilingFuture {
+ private final ScheduledFuture<?> future;
+ private final Runnable handler;
+
+ public ProfilingFuture(long duration, Runnable handler) {
+ this.handler = handler;
+ this.future = scheduledExecutor.schedule(handler, duration, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Verify the profiling request was done or not.
+ *
+ * @return true represents the request was finished.
+ */
+ public boolean isDone() {
+ return this.future == null || this.future.isDone();
+ }
+
+ /**
+ * Try to cancel the Profiling Request. Note that we'll trigger the handler once cancelled
+ * successfully.
+ *
+ * @return true if cancelled, false for failed.
+ */
+ public boolean cancel() {
+ if (isDone()) {
+ return true;
+ }
+ if (!future.cancel(true)) {
+ return false;
+ }
+ // If cancelled, we have to trigger handler immediately for stopping profiler.
+ handler.run();
+ return true;
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
index f6c4a59ca5a..8b2e63e1ec3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
@@ -41,6 +41,9 @@ import org.apache.flink.runtime.rest.handler.cluster.JobManagerCustomLogHandler;
import org.apache.flink.runtime.rest.handler.cluster.JobManagerEnvironmentHandler;
import org.apache.flink.runtime.rest.handler.cluster.JobManagerLogFileHandler;
import org.apache.flink.runtime.rest.handler.cluster.JobManagerLogListHandler;
+import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingFileHandler;
+import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingHandler;
+import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingListHandler;
import org.apache.flink.runtime.rest.handler.cluster.JobManagerThreadDumpHandler;
import org.apache.flink.runtime.rest.handler.cluster.ShutdownHandler;
import org.apache.flink.runtime.rest.handler.dataset.ClusterDataSetDeleteHandlers;
@@ -131,6 +134,9 @@ import org.apache.flink.runtime.rest.messages.checkpoints.TaskCheckpointStatisti
import org.apache.flink.runtime.rest.messages.cluster.JobManagerCustomLogHeaders;
import org.apache.flink.runtime.rest.messages.cluster.JobManagerLogFileHeader;
import org.apache.flink.runtime.rest.messages.cluster.JobManagerLogListHeaders;
+import org.apache.flink.runtime.rest.messages.cluster.JobManagerProfilingFileHeaders;
+import org.apache.flink.runtime.rest.messages.cluster.JobManagerProfilingHeaders;
+import org.apache.flink.runtime.rest.messages.cluster.JobManagerProfilingListHeaders;
import org.apache.flink.runtime.rest.messages.cluster.JobManagerStdoutFileHeader;
import org.apache.flink.runtime.rest.messages.cluster.JobManagerThreadDumpHeaders;
import org.apache.flink.runtime.rest.messages.cluster.ShutdownHeaders;
@@ -958,6 +964,42 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
handlers.add(
Tuple2.of(JobManagerThreadDumpHeaders.getInstance(), jobManagerThreadDumpHandler));
+ // load profiler relative handlers
+ if (clusterConfiguration.get(RestOptions.ENABLE_PROFILER)) {
+ final JobManagerProfilingHandler jobManagerProfilingHandler =
+ new JobManagerProfilingHandler(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ JobManagerProfilingHeaders.getInstance(),
+ clusterConfiguration);
+ final JobManagerProfilingListHandler jobManagerProfilingListHandler =
+ new JobManagerProfilingListHandler(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ JobManagerProfilingListHeaders.getInstance(),
+ clusterConfiguration);
+ final JobManagerProfilingFileHandler jobManagerProfilingFileHandler =
+ new JobManagerProfilingFileHandler(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ JobManagerProfilingFileHeaders.getInstance(),
+ clusterConfiguration);
+ handlers.add(
+ Tuple2.of(
+ JobManagerProfilingHeaders.getInstance(), jobManagerProfilingHandler));
+ handlers.add(
+ Tuple2.of(
+ JobManagerProfilingListHeaders.getInstance(),
+ jobManagerProfilingListHandler));
+ handlers.add(
+ Tuple2.of(
+ JobManagerProfilingFileHeaders.getInstance(),
+ jobManagerProfilingFileHandler));
+ }
+
// TaskManager log and stdout file handler
final Time cacheEntryDuration = Time.milliseconds(restConfiguration.getRefreshInterval());