This is an automated email from the ASF dual-hosted git repository.
jlfsdtc pushed a commit to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
The following commit(s) were added to refs/heads/kylin5 by this push:
new 87753ce44c KYLIN-6128 allow job tracking URL reported from job servers
(#2358)
87753ce44c is described below
commit 87753ce44c2569674c8af0102b998cf35927045a
Author: jlfsdtc <[email protected]>
AuthorDate: Wed Nov 11 10:12:48 2026 +0800
KYLIN-6128 allow job tracking URL reported from job servers (#2358)
---
.../org/apache/kylin/rest/util/SparkUIUtil.java | 23 +++++++++++------
.../apache/kylin/rest/util/SparkUIUtilTest.java | 10 +++++---
.../org/apache/kylin/common/KylinConfigBase.java | 28 ++++++++++++---------
.../kylin/rest/controller/JobController.java | 5 +---
.../kylin/rest/controller/JobControllerTest.java | 29 ++++++++++------------
.../apache/kylin/rest/service/JobInfoService.java | 5 ++++
6 files changed, 57 insertions(+), 43 deletions(-)
diff --git
a/src/common-service/src/main/java/org/apache/kylin/rest/util/SparkUIUtil.java
b/src/common-service/src/main/java/org/apache/kylin/rest/util/SparkUIUtil.java
index 36b562dbe1..edbd03c9e5 100644
---
a/src/common-service/src/main/java/org/apache/kylin/rest/util/SparkUIUtil.java
+++
b/src/common-service/src/main/java/org/apache/kylin/rest/util/SparkUIUtil.java
@@ -24,7 +24,6 @@ import java.net.UnknownHostException;
import java.util.Collections;
import java.util.Locale;
import java.util.Objects;
-
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@@ -34,6 +33,9 @@ import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.exception.KylinRuntimeException;
import org.apache.kylin.common.util.AddressUtil;
+import org.apache.kylin.common.util.HadoopUtil;
+import org.apache.kylin.rest.cluster.ClusterManager;
+import org.apache.kylin.tool.util.HadoopConfExtractor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
@@ -42,7 +44,9 @@ import org.springframework.http.client.ClientHttpRequest;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.springframework.web.util.UriComponentsBuilder;
+import lombok.experimental.UtilityClass;
+@UtilityClass
public class SparkUIUtil {
private static final HttpComponentsClientHttpRequestFactory factory = new
HttpComponentsClientHttpRequestFactory(
@@ -53,9 +57,6 @@ public class SparkUIUtil {
private static final String SPARK_UI_PROXY_HEADER = "X-Kylin-Proxy-Path";
- private SparkUIUtil() {
- }
-
public static void resendSparkUIRequest(HttpServletRequest servletRequest,
HttpServletResponse servletResponse,
String sparkUiUrl, String uriPath, String proxyLocationBase)
throws IOException {
URI originTarget = validateProxyUrl(sparkUiUrl);
@@ -140,7 +141,7 @@ public class SparkUIUtil {
return target;
}
- public static void validateJobTrackingUrl(String url, String requestHost) {
+ public static void validateJobTrackingUrl(String url, String requestHost,
ClusterManager clusterManager) {
URI target = validateProxyUrl(url);
KylinConfig config = KylinConfig.getInstanceFromEnv();
@@ -149,7 +150,7 @@ public class SparkUIUtil {
return;
}
- String yarnStatusUrl = config.getYarnStatusCheckUrl();
+ String yarnStatusUrl =
HadoopConfExtractor.extractYarnMasterUrl(HadoopUtil.getHadoopConfFromSparkEngine());
if (StringUtils.isNotBlank(yarnStatusUrl)) {
URI allowedYarnTarget = validateProxyUrl(yarnStatusUrl);
if (isSameAuthority(target, allowedYarnTarget)) {
@@ -161,7 +162,15 @@ public class SparkUIUtil {
return;
}
- throw new KylinRuntimeException("Tracking URL host is not allowed: " +
target.getHost());
+ boolean matchJobServer =
clusterManager.getJobServers().stream().map(request ->
request.getHost().split(":")[0])
+ .anyMatch(host -> isSameHost(requestHost, host));
+
+ if (matchJobServer) {
+ return;
+ }
+
+ throw new KylinRuntimeException(String.format(Locale.ROOT,
+ "Tracking URL host is not allowed: %s, requestHost: %s",
target.getHost(), requestHost));
}
private static boolean isSameAuthority(URI first, URI second) {
diff --git
a/src/common-service/src/test/java/org/apache/kylin/rest/util/SparkUIUtilTest.java
b/src/common-service/src/test/java/org/apache/kylin/rest/util/SparkUIUtilTest.java
index fefed0122e..87cb6a61cb 100644
---
a/src/common-service/src/test/java/org/apache/kylin/rest/util/SparkUIUtilTest.java
+++
b/src/common-service/src/test/java/org/apache/kylin/rest/util/SparkUIUtilTest.java
@@ -22,10 +22,12 @@ import static org.junit.Assert.assertEquals;
import org.apache.kylin.common.exception.KylinRuntimeException;
import org.apache.kylin.common.util.NLocalFileMetadataTestCase;
+import org.apache.kylin.rest.cluster.ClusterManager;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
public class SparkUIUtilTest extends NLocalFileMetadataTestCase {
@@ -53,12 +55,12 @@ public class SparkUIUtilTest extends
NLocalFileMetadataTestCase {
@Test
public void testValidateJobTrackingUrl() {
- SparkUIUtil.validateJobTrackingUrl("http://127.0.0.1:4040";,
"127.0.0.1");
+ ClusterManager clusterManager = Mockito.mock(ClusterManager.class);
+ SparkUIUtil.validateJobTrackingUrl("http://127.0.0.1:4040";,
"127.0.0.1", clusterManager);
Assert.assertThrows(KylinRuntimeException.class,
- () ->
SparkUIUtil.validateJobTrackingUrl("http://10.0.0.1:4040";, "127.0.0.1"));
+ () ->
SparkUIUtil.validateJobTrackingUrl("http://10.0.0.1:4040";, "127.0.0.1",
clusterManager));
Assert.assertThrows(KylinRuntimeException.class,
- () -> SparkUIUtil.validateJobTrackingUrl("ftp://127.0.0.1";,
"127.0.0.1"));
+ () -> SparkUIUtil.validateJobTrackingUrl("ftp://127.0.0.1";,
"127.0.0.1", clusterManager));
}
}
-
diff --git
a/src/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
b/src/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index ddad0607a3..24505d4766 100644
--- a/src/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/src/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -790,7 +790,7 @@ public abstract class KylinConfigBase implements
Serializable {
* check stays meaningful.
*/
private String getDefaultBroadcastToken() {
- return DigestUtils.sha256Hex("kylin.server.broadcast-token:" +
getMetadataUrl());
+ return DigestUtils.sha256Hex("kylin.server.broadcast-token:" +
getMetadataUrlPrefix());
}
public String getServerHttpsKeyAlias() {
@@ -4549,12 +4549,13 @@ public abstract class KylinConfigBase implements
Serializable {
}
public Set<String> getSourceJdbcWhiteListSchemes() {
- String config =
StringUtils.deleteWhitespace(getOptional("kylin.source.jdbc.white-list.schemes",
- "h2,mysql,postgresql,sqlserver,snowflake"));
+ String config = StringUtils.deleteWhitespace(
+ getOptional("kylin.source.jdbc.white-list.schemes",
"h2,mysql,postgresql,sqlserver,snowflake"));
if (StringUtils.isBlank(config)) {
return Collections.emptySet();
}
- return Arrays.stream(config.split(",")).map(value ->
value.toLowerCase(Locale.ROOT)).collect(Collectors.toSet());
+ return Arrays.stream(config.split(",")).map(value ->
value.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toSet());
}
public String getSourceJdbcWhiteListValidatorClassByScheme(String scheme) {
@@ -4571,8 +4572,8 @@ public abstract class KylinConfigBase implements
Serializable {
if (!whiteListSchemes.contains(scheme)) {
return Collections.emptySet();
}
- String config =
StringUtils.deleteWhitespace(getOptional(String.format(Locale.ROOT,
- "kylin.source.jdbc.white-list.%s.url-param-keys", scheme),
""));
+ String config = StringUtils.deleteWhitespace(
+ getOptional(String.format(Locale.ROOT,
"kylin.source.jdbc.white-list.%s.url-param-keys", scheme), ""));
if (StringUtils.isBlank(config)) {
return Collections.emptySet();
}
@@ -4584,8 +4585,9 @@ public abstract class KylinConfigBase implements
Serializable {
if (!whiteListSchemes.contains(scheme)) {
return Collections.emptySet();
}
- String config =
StringUtils.deleteWhitespace(getOptional(String.format(Locale.ROOT,
- "kylin.source.jdbc.white-list.%s.drivers", scheme),
getDefaultSourceJdbcDrivers(scheme)));
+ String config = StringUtils.deleteWhitespace(
+ getOptional(String.format(Locale.ROOT,
"kylin.source.jdbc.white-list.%s.drivers", scheme),
+ getDefaultSourceJdbcDrivers(scheme)));
if (StringUtils.isBlank(config)) {
return Collections.emptySet();
}
@@ -4609,8 +4611,9 @@ public abstract class KylinConfigBase implements
Serializable {
defaultValue = "";
break;
}
- String config =
StringUtils.deleteWhitespace(getOptional(String.format(Locale.ROOT,
- "kylin.source.jdbc.white-list.%s.semicolon-param-keys",
scheme), defaultValue));
+ String config = StringUtils.deleteWhitespace(
+ getOptional(String.format(Locale.ROOT,
"kylin.source.jdbc.white-list.%s.semicolon-param-keys", scheme),
+ defaultValue));
if (StringUtils.isBlank(config)) {
return Collections.emptySet();
}
@@ -4631,8 +4634,9 @@ public abstract class KylinConfigBase implements
Serializable {
defaultValue = "";
break;
}
- String config =
StringUtils.deleteWhitespace(getOptional(String.format(Locale.ROOT,
- "kylin.source.jdbc.white-list.%s.parenthesis-param-keys",
scheme), defaultValue));
+ String config = StringUtils.deleteWhitespace(getOptional(
+ String.format(Locale.ROOT,
"kylin.source.jdbc.white-list.%s.parenthesis-param-keys", scheme),
+ defaultValue));
if (StringUtils.isBlank(config)) {
return Collections.emptySet();
}
diff --git
a/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/JobController.java
b/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/JobController.java
index 3234d5a0be..d6aa21aa4d 100644
---
a/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/JobController.java
+++
b/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/JobController.java
@@ -30,7 +30,6 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
-
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@@ -64,7 +63,6 @@ import org.apache.kylin.rest.service.JobInfoService;
import org.apache.kylin.rest.service.JobService;
import org.apache.kylin.rest.service.RouteService;
import org.apache.kylin.rest.util.AclEvaluate;
-import org.apache.kylin.rest.util.SparkUIUtil;
import org.apache.kylin.utils.GlutenCacheUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -83,7 +81,6 @@ import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
-
import io.swagger.annotations.ApiOperation;
import lombok.val;
@@ -325,7 +322,7 @@ public class JobController extends BaseController {
InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest);
checkProjectName(sparkJobUpdateRequest.getProject());
if (StringUtils.isNotBlank(sparkJobUpdateRequest.getYarnAppUrl())) {
-
SparkUIUtil.validateJobTrackingUrl(sparkJobUpdateRequest.getYarnAppUrl(),
+
jobInfoService.validateJobTrackingUrl(sparkJobUpdateRequest.getYarnAppUrl(),
servletRequest.getRemoteAddr());
}
return updateJobInfoWithCheck(sparkJobUpdateRequest.getProject(),
sparkJobUpdateRequest.getJobId(),
diff --git
a/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java
b/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java
index ead614d1e9..2b7db50098 100644
---
a/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java
+++
b/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java
@@ -28,6 +28,7 @@ import java.util.stream.Collectors;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.exception.KylinException;
+import org.apache.kylin.common.exception.KylinRuntimeException;
import
org.apache.kylin.common.persistence.transaction.BroadcastEventReadyNotifier;
import org.apache.kylin.common.util.JsonUtil;
import org.apache.kylin.common.util.NLocalFileMetadataTestCase;
@@ -342,14 +343,11 @@ public class JobControllerTest extends
NLocalFileMetadataTestCase {
request.setMemory("1024");
request.setQueueName("queue");
Mockito.doNothing().when(jobInfoService).updateSparkJobInfo(request);
- MvcResult result = mockMvc
-
.perform(MockMvcRequestBuilders.put("/api/jobs/spark").contentType(MediaType.APPLICATION_JSON)
- .content(JsonUtil.writeValueAsString(request))
- .with(req -> {
- req.setRemoteAddr("127.0.0.1");
- return req;
- })
-
.accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON)))
+ MvcResult result =
mockMvc.perform(MockMvcRequestBuilders.put("/api/jobs/spark")
+
.contentType(MediaType.APPLICATION_JSON).content(JsonUtil.writeValueAsString(request)).with(req
-> {
+ req.setRemoteAddr("127.0.0.1");
+ return req;
+
}).accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON)))
.andExpect(MockMvcResultMatchers.status().isOk()).andReturn();
Map<String, String> response =
JsonUtil.readValueAsMap(result.getResponse().getContentAsString());
Assert.assertEquals(response.get("code"), KylinException.CODE_SUCCESS);
@@ -367,13 +365,14 @@ public class JobControllerTest extends
NLocalFileMetadataTestCase {
request.setTaskId(job.getId() + "_00_00");
request.setYarnAppUrl("http://10.0.0.1:4040";);
+ Mockito.doThrow(new KylinRuntimeException("Tracking URL host is not
allowed")).when(jobInfoService)
+ .validateJobTrackingUrl(request.getYarnAppUrl(), "127.0.0.1");
+
mockMvc.perform(MockMvcRequestBuilders.put("/api/jobs/spark").contentType(MediaType.APPLICATION_JSON)
- .content(JsonUtil.writeValueAsString(request))
- .with(req -> {
+ .content(JsonUtil.writeValueAsString(request)).with(req -> {
req.setRemoteAddr("127.0.0.1");
return req;
- })
- .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON)))
+
}).accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON)))
.andExpect(MockMvcResultMatchers.status().isInternalServerError());
Mockito.verify(jobInfoService,
Mockito.never()).updateSparkJobInfo(Mockito.any());
@@ -574,8 +573,7 @@ public class JobControllerTest extends
NLocalFileMetadataTestCase {
MvcResult result = mockMvc
.perform(MockMvcRequestBuilders.put("/api/jobs/error").contentType(MediaType.APPLICATION_JSON)
-
.header(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER,
- getTestConfig().getBroadcastToken())
+
.header(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER,
getTestConfig().getBroadcastToken())
.content(JsonUtil.writeValueAsString(request))
.accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON)))
.andExpect(MockMvcResultMatchers.status().isOk()).andReturn();
@@ -686,8 +684,7 @@ public class JobControllerTest extends
NLocalFileMetadataTestCase {
@Test
public void routeGlutenCache() throws Exception {
val servletRequest = new MockHttpServletRequest();
- val request = new LoadGlutenCacheRequest("default",
- Lists.newArrayList("CACHE DATA SELECT * FROM 'test-path'"));
+ val request = new LoadGlutenCacheRequest("default",
Lists.newArrayList("CACHE DATA SELECT * FROM 'test-path'"));
Mockito.when(routeService.routeGlutenCache(request.getCacheCommands(),
servletRequest)).thenReturn(true);
val mvcResult =
mockMvc.perform(MockMvcRequestBuilders.post("/api/jobs/gluten_cache") //
.contentType(MediaType.APPLICATION_JSON).content(JsonUtil.writeValueAsString(request)))
diff --git
a/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobInfoService.java
b/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobInfoService.java
index 32953f9bf4..68efbb2d1e 100644
---
a/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobInfoService.java
+++
b/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobInfoService.java
@@ -110,6 +110,7 @@ import org.apache.kylin.rest.util.AclEvaluate;
import org.apache.kylin.rest.util.JobDriverUIUtil;
import org.apache.kylin.rest.util.JobFilterUtil;
import org.apache.kylin.rest.util.SparkHistoryUIUtil;
+import org.apache.kylin.rest.util.SparkUIUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -560,6 +561,10 @@ public class JobInfoService extends BasicService
implements JobSupporter {
return ExecutableState.valueOf(status);
}
+ public void validateJobTrackingUrl(String url, String requestHost) {
+ SparkUIUtil.validateJobTrackingUrl(url, requestHost, clusterManager);
+ }
+
/**
* update the spark job info, such as yarnAppId, yarnAppUrl.
*
