[FLINK-7694][flip6] Migrate JobMetricsHandler to new RestServerEndpoint

Migrate logic in
org.apache.flink.runtime.rest.handler.legacy.metrics.JobMetrisHandler to new
handler, and add new handler to DispatcherRestEndpoint.

[FLINK-7694][flip6] Use jobid path parameter constant in JobMetricsHandlerTest

This closes #5084.

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java

@@ -46,6 +46,7 @@ import org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatsCach
 import org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointingStatisticsHandler;
 import org.apache.flink.runtime.rest.handler.job.checkpoints.TaskCheckpointStatisticDetailsHandler;
 import org.apache.flink.runtime.rest.handler.job.metrics.JobManagerMetricsHandler;
+import org.apache.flink.runtime.rest.handler.job.metrics.JobMetricsHandler;
 import org.apache.flink.runtime.rest.handler.job.metrics.JobVertexMetricsHandler;
 import org.apache.flink.runtime.rest.handler.job.metrics.SubtaskMetricsHandler;
 import org.apache.flink.runtime.rest.handler.job.metrics.TaskManagerMetricsHandler;
@@ -72,6 +73,7 @@ import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistic
 import org.apache.flink.runtime.rest.messages.checkpoints.TaskCheckpointStatisticsHeaders;
 import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
 import org.apache.flink.runtime.rest.messages.job.metrics.JobManagerMetricsHeaders;
+import org.apache.flink.runtime.rest.messages.job.metrics.JobMetricsHeaders;
 import org.apache.flink.runtime.rest.messages.job.metrics.JobVertexMetricsHeaders;
 import org.apache.flink.runtime.rest.messages.job.metrics.SubtaskMetricsHeaders;
 import org.apache.flink.runtime.rest.messages.job.metrics.TaskManagerMetricsHeaders;
@@ -321,6 +323,13 @@ public class DispatcherRestEndpoint extends RestServerEndpoint {
 			responseHeaders,
 			metricFetcher);
 
+		final JobMetricsHandler jobMetricsHandler = new JobMetricsHandler(
+			restAddressFuture,
+			leaderRetriever,
+			timeout,
+			responseHeaders,
+			metricFetcher);
+
 		final SubtaskMetricsHandler subtaskMetricsHandler = new SubtaskMetricsHandler(
 			restAddressFuture,
 			leaderRetriever,
@@ -378,6 +387,7 @@ public class DispatcherRestEndpoint extends RestServerEndpoint {
 		handlers.add(Tuple2.of(TaskManagerDetailsHeaders.getInstance(), taskManagerDetailsHandler));
 		handlers.add(Tuple2.of(SubtasksTimesHeaders.getInstance(), subtasksTimesHandler));
 		handlers.add(Tuple2.of(JobVertexMetricsHeaders.getInstance(), jobVertexMetricsHandler));
+		handlers.add(Tuple2.of(JobMetricsHeaders.getInstance(), jobMetricsHandler));
 		handlers.add(Tuple2.of(SubtaskMetricsHeaders.getInstance(), subtaskMetricsHandler));
 		handlers.add(Tuple2.of(TaskManagerMetricsHeaders.getInstance(), taskManagerMetricsHandler));
 		handlers.add(Tuple2.of(JobManagerMetricsHeaders.getInstance(), jobManagerMetricsHandler));

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/JobMetricsHandler.java

+/*
+ * 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.job.metrics;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.dispatcher.DispatcherGateway;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcher;
+import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.job.metrics.JobMetricsHeaders;
+import org.apache.flink.runtime.rest.messages.job.metrics.JobMetricsMessageParameters;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Request handler that returns for a given job a list of all available metrics or the values for a set of metrics.
+ */
+public class JobMetricsHandler extends AbstractMetricsHandler<JobMetricsMessageParameters> {
+
+	public JobMetricsHandler(
+			final CompletableFuture<String> localRestAddress,
+			final GatewayRetriever<DispatcherGateway> leaderRetriever,
+			final Time timeout,
+			final Map<String, String> headers,
+			final MetricFetcher metricFetcher) {
+		super(localRestAddress, leaderRetriever, timeout, headers, JobMetricsHeaders.getInstance(),
+			metricFetcher);
+	}
+
+	@Nullable
+	@Override
+	protected MetricStore.ComponentMetricStore getComponentMetricStore(
+		final HandlerRequest<EmptyRequestBody, JobMetricsMessageParameters> request,
+		final MetricStore metricStore) {
+		return metricStore.getJobMetricStore(request.getPathParameter(JobIDPathParameter.class).toString());
+	}
+
+}

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobMetricsHeaders.java

+/*
+ * 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.job.metrics;
+
+import org.apache.flink.runtime.rest.handler.job.metrics.JobMetricsHandler;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+
+/**
+ * {@link MessageHeaders} for {@link JobMetricsHandler}.
+ */
+public final class JobMetricsHeaders extends AbstractMetricsHeaders<JobMetricsMessageParameters> {
+
+	private static final JobMetricsHeaders INSTANCE = new JobMetricsHeaders();
+
+	private JobMetricsHeaders() {
+	}
+
+	@Override
+	public JobMetricsMessageParameters getUnresolvedMessageParameters() {
+		return new JobMetricsMessageParameters();
+	}
+
+	@Override
+	public String getTargetRestEndpointURL() {
+		return "/jobs/:" + JobIDPathParameter.KEY + "/metrics";
+	}
+
+	public static JobMetricsHeaders getInstance() {
+		return INSTANCE;
+	}
+
+}

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobMetricsMessageParameters.java

+/*
+ * 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.job.metrics;
+
+import org.apache.flink.runtime.rest.handler.job.metrics.JobMetricsHandler;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageQueryParameter;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * {@link MessageParameters} for {@link JobMetricsHandler}.
+ */
+public class JobMetricsMessageParameters extends JobMessageParameters {
+
+	private final MetricsFilterParameter metricsFilterParameter = new MetricsFilterParameter();
+
+	@Override
+	public Collection<MessageQueryParameter<?>> getQueryParameters() {
+		return Collections.singletonList(metricsFilterParameter);
+	}
+
+}

flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/JobMetricsHandlerTest.java

+/*
+ * 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.job.metrics;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.metrics.dump.QueryScopeInfo;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for {@link JobMetricsHandler}.
+ */
+public class JobMetricsHandlerTest extends MetricsHandlerTestBase<JobMetricsHandler> {
+
+	private static final String TEST_JOB_ID = new JobID().toString();
+
+	@Override
+	JobMetricsHandler getMetricsHandler() {
+		return new JobMetricsHandler(
+			TEST_REST_ADDRESS,
+			leaderRetriever,
+			TIMEOUT,
+			TEST_HEADERS,
+			mockMetricFetcher);
+	}
+
+	@Override
+	QueryScopeInfo getQueryScopeInfo() {
+		return new QueryScopeInfo.JobQueryScopeInfo(TEST_JOB_ID);
+	}
+
+	@Override
+	Map<String, String> getPathParameters() {
+		Map<String, String> pathParameters = new HashMap<>();
+		pathParameters.put(JobIDPathParameter.KEY, TEST_JOB_ID);
+		return pathParameters;
+	}
+
+}

flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/metrics/JobMetricsHeadersTest.java

+/*
+ * 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.job.metrics;
+
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+
+import org.junit.Test;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link JobMetricsHeaders}.
+ */
+public class JobMetricsHeadersTest {
+
+	private final JobMetricsHeaders jobMetricsHeaders = JobMetricsHeaders.getInstance();
+
+	@Test
+	public void testUrl() {
+		assertThat(jobMetricsHeaders.getTargetRestEndpointURL(),
+			equalTo("/jobs/:" + JobIDPathParameter.KEY + "/metrics"));
+	}
+
+	@Test
+	public void testMessageParameters() {
+		assertThat(jobMetricsHeaders.getUnresolvedMessageParameters(), instanceOf
+			(JobMetricsMessageParameters.class));
+	}
+
+}