This is an automated email from the ASF dual-hosted git repository. aiceflower pushed a commit to branch release-0.9.4 in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 3ffd280d9caad0430f8f920257ce3488c764cd8b Author: cooperyang <[email protected]> AuthorDate: Thu Jun 4 15:25:01 2020 +0800 adjust gateway and add a ha module to adapt context service (#436) adjust gateway and add a ha module to adapt context service --- .../cs/listener/test/TestListenerManager.java | 1 - .../config/GatewaySpringConfiguration.scala | 2 +- .../linkis/gateway/http/GatewayContext.scala | 9 ++ .../linkis/gateway/security/GatewaySSOUtils.scala | 2 +- .../linkis/gateway/security/UserRestful.scala | 4 +- gateway/gateway-httpclient-support/pom.xml | 2 +- .../linkis/httpclient/dws/DWSHttpClient.scala | 21 ++-- .../StaticAuthenticationStrategy.scala | 4 +- .../TokenAuthenticationStrategy.scala | 7 +- .../dws/config/DWSClientConfigBuilder.scala | 2 +- .../dws/discovery/DWSGatewayDiscovery.scala | 4 +- .../dws/response/DWSAuthenticationResult.scala | 32 ++++-- .../dws/response/DWSHeartbeatResult.scala | 25 ++++- gateway/gateway-ujes-support/pom.xml | 6 + .../ujes/parser/EntranceRequestGatewayParser.scala | 3 +- .../ujes/route/HaContextGatewayRouter.scala | 124 +++++++++++++++++++++ .../route/contextservice/ContextIdParserImpl.scala | 52 +++++++++ .../SpringCloudGatewayConfiguration.scala | 1 - 18 files changed, 263 insertions(+), 38 deletions(-) diff --git a/contextservice/cs-listener/src/test/java/com/webank/wedatasphere/linkis/cs/listener/test/TestListenerManager.java b/contextservice/cs-listener/src/test/java/com/webank/wedatasphere/linkis/cs/listener/test/TestListenerManager.java index 3835c672d8..c86d8a4c85 100644 --- a/contextservice/cs-listener/src/test/java/com/webank/wedatasphere/linkis/cs/listener/test/TestListenerManager.java +++ b/contextservice/cs-listener/src/test/java/com/webank/wedatasphere/linkis/cs/listener/test/TestListenerManager.java @@ -7,7 +7,6 @@ import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKey; import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKeyValue; import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextValue; import com.webank.wedatasphere.linkis.cs.listener.ListenerBus.ContextAsyncListenerBus; -import com.webank.wedatasphere.linkis.cs.listener.callback.imp.ClientSource; import com.webank.wedatasphere.linkis.cs.listener.callback.imp.ContextKeyValueBean; import com.webank.wedatasphere.linkis.cs.listener.callback.imp.DefaultContextIDCallbackEngine; import com.webank.wedatasphere.linkis.cs.listener.callback.imp.DefaultContextKeyCallbackEngine; diff --git a/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/config/GatewaySpringConfiguration.scala b/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/config/GatewaySpringConfiguration.scala index 67eab5ff64..f8ef9a71e9 100644 --- a/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/config/GatewaySpringConfiguration.scala +++ b/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/config/GatewaySpringConfiguration.scala @@ -16,7 +16,7 @@ package com.webank.wedatasphere.linkis.gateway.config -import com.webank.wedatasphere.linkis.gateway.security.{LDAPUserRestful, SecurityHook, SecurityFilter, UserRestful} +import com.webank.wedatasphere.linkis.gateway.security.{LDAPUserRestful, SecurityFilter, SecurityHook, UserRestful} import javax.annotation.PostConstruct import org.springframework.beans.factory.annotation.Autowired import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean diff --git a/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/http/GatewayContext.scala b/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/http/GatewayContext.scala index 87906ee5ee..092ece9e1e 100644 --- a/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/http/GatewayContext.scala +++ b/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/http/GatewayContext.scala @@ -16,6 +16,10 @@ package com.webank.wedatasphere.linkis.gateway.http +import java.util + +import com.webank.wedatasphere.linkis.server.JMap + /** * created by cooperyang on 2019/1/9. */ @@ -32,6 +36,7 @@ trait GatewayContext { def setGatewayRoute(gatewayRoute: GatewayRoute): Unit def getGatewayRoute: GatewayRoute + def getParams: JMap[String, String] } class BaseGatewayContext extends GatewayContext { private var request: GatewayHttpRequest = _ @@ -39,6 +44,8 @@ class BaseGatewayContext extends GatewayContext { private var webSocketRequest: Boolean = false private var gatewayRoute: GatewayRoute = _ + private val props: JMap[String, String] = new util.HashMap[String, String]() + override def getRequest: GatewayHttpRequest = request override def setRequest(request: GatewayHttpRequest): Unit = this.request = request @@ -54,4 +61,6 @@ class BaseGatewayContext extends GatewayContext { override def setGatewayRoute(gatewayRoute: GatewayRoute): Unit = this.gatewayRoute = gatewayRoute override def getGatewayRoute: GatewayRoute = gatewayRoute + + override def getParams: JMap[String, String] = this.props } \ No newline at end of file diff --git a/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/security/GatewaySSOUtils.scala b/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/security/GatewaySSOUtils.scala index 2d459d5228..08dd795913 100644 --- a/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/security/GatewaySSOUtils.scala +++ b/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/security/GatewaySSOUtils.scala @@ -19,8 +19,8 @@ package com.webank.wedatasphere.linkis.gateway.security import com.webank.wedatasphere.linkis.common.utils.{Logging, Utils} import com.webank.wedatasphere.linkis.gateway.http.{GatewayContext, GatewayHttpRequest} import com.webank.wedatasphere.linkis.server.exception.LoginExpireException -import com.webank.wedatasphere.linkis.server.security.{SSOUtils, ServerSSOUtils} import com.webank.wedatasphere.linkis.server.security.SecurityFilter._ +import com.webank.wedatasphere.linkis.server.security.{SSOUtils, ServerSSOUtils} import javax.servlet.http.Cookie import scala.collection.JavaConversions._ diff --git a/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/security/UserRestful.scala b/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/security/UserRestful.scala index 8678f531ef..d01b9dbf07 100644 --- a/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/security/UserRestful.scala +++ b/gateway/core/src/main/scala/com/webank/wedatasphere/linkis/gateway/security/UserRestful.scala @@ -16,9 +16,7 @@ package com.webank.wedatasphere.linkis.gateway.security -import java.util.Random - -import com.google.gson.{Gson, JsonObject} +import com.google.gson.Gson import com.webank.wedatasphere.linkis.common.utils.{Logging, RSAUtils, Utils} import com.webank.wedatasphere.linkis.gateway.config.GatewayConfiguration import com.webank.wedatasphere.linkis.gateway.http.GatewayContext diff --git a/gateway/gateway-httpclient-support/pom.xml b/gateway/gateway-httpclient-support/pom.xml index dce31146f9..c0daf49b0e 100644 --- a/gateway/gateway-httpclient-support/pom.xml +++ b/gateway/gateway-httpclient-support/pom.xml @@ -42,7 +42,7 @@ <dependency> <groupId>commons-beanutils</groupId> <artifactId>commons-beanutils</artifactId> - <version>1.7.0</version> + <version>1.9.4</version> </dependency> <dependency> <groupId>org.reflections</groupId> diff --git a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/DWSHttpClient.scala b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/DWSHttpClient.scala index 6f9c481e4f..d36005c115 100644 --- a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/DWSHttpClient.scala +++ b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/DWSHttpClient.scala @@ -24,7 +24,6 @@ import java.text.SimpleDateFormat import java.util import com.fasterxml.jackson.databind.ObjectMapper -import com.ning.http.client.Response import com.webank.wedatasphere.linkis.common.io.{Fs, FsPath} import com.webank.wedatasphere.linkis.httpclient.AbstractHttpClient import com.webank.wedatasphere.linkis.httpclient.discovery.Discovery @@ -37,6 +36,7 @@ import com.webank.wedatasphere.linkis.httpclient.response.{HttpResult, ListResul import com.webank.wedatasphere.linkis.storage.FSFactory import org.apache.commons.beanutils.BeanUtils import org.apache.commons.lang.ClassUtils +import org.apache.http.HttpResponse import scala.collection.JavaConversions.mapAsJavaMap @@ -44,7 +44,7 @@ import scala.collection.JavaConversions.mapAsJavaMap * created by cooperyang on 2019/5/20. */ class DWSHttpClient(clientConfig: DWSClientConfig, clientName: String) - extends AbstractHttpClient(clientConfig, clientName) { + extends AbstractHttpClient(clientConfig, clientName) { override protected def createDiscovery(): Discovery = new DWSGatewayDiscovery @@ -57,21 +57,25 @@ class DWSHttpClient(clientConfig: DWSClientConfig, clientName: String) requestAction } - override protected def httpResponseToResult(response: Response, requestAction: HttpAction): Option[Result] = { - val url = requestAction.getURL + override protected def httpResponseToResult(response: HttpResponse, requestAction: HttpAction, responseBody: String): Option[Result] = { + var entity = response.getEntity + val statusCode: Int = response.getStatusLine.getStatusCode + val url: String = requestAction.getURL + val contentType: String = entity.getContentType.getValue DWSHttpMessageFactory.getDWSHttpMessageResult(url).map { case DWSHttpMessageResultInfo(_, clazz) => clazz match { case c if ClassUtils.isAssignable(c, classOf[DWSResult]) => val dwsResult = clazz.getConstructor().newInstance().asInstanceOf[DWSResult] - dwsResult.set(response.getResponseBody, response.getStatusCode, response.getUri.toString, response.getContentType) + dwsResult.set(responseBody, statusCode, url, contentType) BeanUtils.populate(dwsResult, dwsResult.getData) return Some(dwsResult) case _ => } + def transfer(value: Result, map: Map[String, Object]): Unit = { value match { case httpResult: HttpResult => - httpResult.set(response.getResponseBody, response.getStatusCode, response.getUri.toString, response.getContentType) + httpResult.set(responseBody, statusCode, url, contentType) case _ => } val javaMap = mapAsJavaMap(map) @@ -89,17 +93,18 @@ class DWSHttpClient(clientConfig: DWSClientConfig, clientName: String) transfer(value, map) value }.toArray - new ListResult(response.getResponseBody, results) + new ListResult(responseBody, results) } }.orElse(nonDWSResponseToResult(response, requestAction)) } - protected def nonDWSResponseToResult(response: Response, requestAction: HttpAction): Option[Result] = None + protected def nonDWSResponseToResult(response: HttpResponse, requestAction: HttpAction): Option[Result] = None protected def fillResultFields(responseMap: util.Map[String, Object], value: Result): Unit = {} //TODO Consistent with workspace, plus expiration time(与workspace保持一致，加上过期时间) override protected def getFsByUser(user: String, path: FsPath): Fs = FSFactory.getFsByProxyUser(path, user) + } object DWSHttpClient { val jacksonJson = new ObjectMapper().setDateFormat(new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ")) diff --git a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/authentication/StaticAuthenticationStrategy.scala b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/authentication/StaticAuthenticationStrategy.scala index ab967b9558..129a2a0f77 100644 --- a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/authentication/StaticAuthenticationStrategy.scala +++ b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/authentication/StaticAuthenticationStrategy.scala @@ -16,7 +16,6 @@ package com.webank.wedatasphere.linkis.httpclient.dws.authentication -import com.ning.http.client.Response import com.webank.wedatasphere.linkis.common.utils.ByteTimeUtils import com.webank.wedatasphere.linkis.httpclient.authentication.{AbstractAuthenticationStrategy, AuthenticationAction, AuthenticationResult} import com.webank.wedatasphere.linkis.httpclient.dws.exception.AuthenticationFailedException @@ -24,6 +23,7 @@ import com.webank.wedatasphere.linkis.httpclient.dws.request.DWSAuthenticationAc import com.webank.wedatasphere.linkis.httpclient.dws.response.DWSAuthenticationResult import com.webank.wedatasphere.linkis.httpclient.request.{Action, UserAction, UserPwdAction} import org.apache.commons.lang.StringUtils +import org.apache.http.HttpResponse /** * created by cooperyang on 2019/5/22. @@ -51,7 +51,7 @@ class StaticAuthenticationStrategy(override protected val sessionMaxAliveTime: L action } - override def getAuthenticationResult(response: Response, requestAction: AuthenticationAction): AuthenticationResult = { + override def getAuthenticationResult(response: HttpResponse, requestAction: AuthenticationAction): AuthenticationResult = { new DWSAuthenticationResult(response, requestAction.serverUrl) } } \ No newline at end of file diff --git a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/authentication/TokenAuthenticationStrategy.scala b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/authentication/TokenAuthenticationStrategy.scala index ac74199fd0..061bcc27f0 100644 --- a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/authentication/TokenAuthenticationStrategy.scala +++ b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/authentication/TokenAuthenticationStrategy.scala @@ -18,11 +18,11 @@ package com.webank.wedatasphere.linkis.httpclient.dws.authentication import java.util -import com.ning.http.client.Response -import com.ning.http.client.cookie.Cookie import com.webank.wedatasphere.linkis.httpclient.authentication._ import com.webank.wedatasphere.linkis.httpclient.dws.exception.AuthenticationFailedException import com.webank.wedatasphere.linkis.httpclient.request.{Action, UserAction} +import org.apache.http.HttpResponse +import org.apache.http.cookie.Cookie /** * Created by enjoyyin on 2019/10/4. @@ -35,6 +35,7 @@ class TokenAuthenticationStrategy(override protected val sessionMaxAliveTime: Lo case _: AuthenticationAction => null case action: UserAction => new HttpAuthentication { import TokenAuthenticationStrategy._ + import scala.collection.JavaConversions._ override def authToCookies: Array[Cookie] = Array.empty @@ -51,7 +52,7 @@ class TokenAuthenticationStrategy(override protected val sessionMaxAliveTime: Lo override protected def getAuthenticationAction(requestAction: Action, serverUrl: String): AuthenticationAction = null - override def getAuthenticationResult(response: Response, requestAction: AuthenticationAction): AuthenticationResult = null + override def getAuthenticationResult(response: HttpResponse, requestAction: AuthenticationAction): AuthenticationResult = null } object TokenAuthenticationStrategy { diff --git a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/config/DWSClientConfigBuilder.scala b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/config/DWSClientConfigBuilder.scala index 51481c7108..2498b8e522 100644 --- a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/config/DWSClientConfigBuilder.scala +++ b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/config/DWSClientConfigBuilder.scala @@ -16,7 +16,7 @@ package com.webank.wedatasphere.linkis.httpclient.dws.config -import com.webank.wedatasphere.linkis.httpclient.config.{ClientConfig, ClientConfigBuilder} +import com.webank.wedatasphere.linkis.httpclient.config.ClientConfigBuilder import com.webank.wedatasphere.linkis.httpclient.dws.exception.UnknownVersionException import org.apache.commons.lang.StringUtils diff --git a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/discovery/DWSGatewayDiscovery.scala b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/discovery/DWSGatewayDiscovery.scala index 2dc2992401..bb77b5beca 100644 --- a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/discovery/DWSGatewayDiscovery.scala +++ b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/discovery/DWSGatewayDiscovery.scala @@ -20,10 +20,10 @@ package com.webank.wedatasphere.linkis.httpclient.dws.discovery -import com.ning.http.client.Response import com.webank.wedatasphere.linkis.httpclient.discovery.{AbstractDiscovery, HeartbeatAction, HeartbeatResult} import com.webank.wedatasphere.linkis.httpclient.dws.request.DWSHeartbeatAction import com.webank.wedatasphere.linkis.httpclient.dws.response.DWSHeartbeatResult +import org.apache.http.HttpResponse import scala.util.Random @@ -43,7 +43,7 @@ class DWSGatewayDiscovery extends AbstractDiscovery { override protected def getHeartbeatAction(serverUrl: String): HeartbeatAction = new DWSHeartbeatAction(serverUrl) - override def getHeartbeatResult(response: Response, requestAction: HeartbeatAction): HeartbeatResult = requestAction match { + override def getHeartbeatResult(response: HttpResponse, requestAction: HeartbeatAction): HeartbeatResult = requestAction match { case h: DWSHeartbeatAction => new DWSHeartbeatResult(response, h.serverUrl) } } diff --git a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/response/DWSAuthenticationResult.scala b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/response/DWSAuthenticationResult.scala index de939f884a..bc2c479b0c 100644 --- a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/response/DWSAuthenticationResult.scala +++ b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/response/DWSAuthenticationResult.scala @@ -18,22 +18,37 @@ package com.webank.wedatasphere.linkis.httpclient.dws.response import java.util -import com.ning.http.client.Response -import com.ning.http.client.cookie.Cookie import com.webank.wedatasphere.linkis.httpclient.authentication.{Authentication, AuthenticationResult, HttpAuthentication} import com.webank.wedatasphere.linkis.httpclient.exception.HttpMessageParseException - -import scala.collection.JavaConversions +import org.apache.http.HttpResponse +import org.apache.http.cookie.Cookie +import org.apache.http.util.EntityUtils /** * created by cooperyang on 2019/5/22. */ -class DWSAuthenticationResult(response: Response, serverUrl: String) extends AuthenticationResult with DWSResult { +class DWSAuthenticationResult(response: HttpResponse, serverUrl: String) extends AuthenticationResult with DWSResult { + + setResponse() + + private def setResponse(): Unit = { + val entity = response.getEntity + val responseBody: String = if (entity != null) { + EntityUtils.toString(entity, "UTF-8") + } else { + null + } + val statusCode: Int = response.getStatusLine.getStatusCode + val url: String = serverUrl + val contentType: String = entity.getContentType.getValue + set(responseBody, statusCode, url, contentType) + } - set(response.getResponseBody, response.getStatusCode, response.getUri.toString, response.getContentType) - override def getAuthentication: Authentication = if(getStatus == 0) new HttpAuthentication { + + override def getAuthentication: Authentication = if (getStatus == 0) new HttpAuthentication { private var lastAccessTime: Long = System.currentTimeMillis - override def authToCookies: Array[Cookie] = JavaConversions.asScalaBuffer(response.getCookies).toArray + + override def authToCookies: Array[Cookie] = Array.empty override def authToHeaders: util.Map[String, String] = new util.HashMap[String, String]() @@ -44,4 +59,5 @@ class DWSAuthenticationResult(response: Response, serverUrl: String) extends Aut override def updateLastAccessTime(): Unit = lastAccessTime = System.currentTimeMillis } else throw new HttpMessageParseException(s"login to gateway $serverUrl failed! Reason: " + getMessage) + } diff --git a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/response/DWSHeartbeatResult.scala b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/response/DWSHeartbeatResult.scala index 3697e33cd6..4030a5a05e 100644 --- a/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/response/DWSHeartbeatResult.scala +++ b/gateway/gateway-httpclient-support/src/main/scala/com/webank/wedatasphere/linkis/httpclient/dws/response/DWSHeartbeatResult.scala @@ -18,16 +18,33 @@ package com.webank.wedatasphere.linkis.httpclient.dws.response import java.util -import com.ning.http.client.Response import com.webank.wedatasphere.linkis.httpclient.discovery.HeartbeatResult +import org.apache.http.HttpResponse +import org.apache.http.util.EntityUtils /** * created by cooperyang on 2019/5/22. */ -class DWSHeartbeatResult(response: Response, serverUrl: String) extends HeartbeatResult with DWSResult { +class DWSHeartbeatResult(response: HttpResponse, serverUrl: String) extends HeartbeatResult with DWSResult { - set(response.getResponseBody, response.getStatusCode, response.getUri.toString, response.getContentType) - if(getStatus != 0) warn(s"heartbeat to gateway $serverUrl failed! message: $getMessage.") + + setResponse() + + private def setResponse(): Unit = { + val entity = response.getEntity + val responseBody: String = if (entity != null) { + EntityUtils.toString(entity, "UTF-8") + } else { + null + } + val statusCode: Int = response.getStatusLine.getStatusCode + val url: String = serverUrl + val contentType: String = entity.getContentType.getValue + set(responseBody, statusCode, url, contentType) + } + + + if (getStatus != 0) warn(s"heartbeat to gateway $serverUrl failed! message: $getMessage.") override val isHealthy: Boolean = getData.get("isHealthy") match { case b: java.lang.Boolean => b case s if s != null => s.toString.toBoolean diff --git a/gateway/gateway-ujes-support/pom.xml b/gateway/gateway-ujes-support/pom.xml index 9a44a17cac..bc041956e6 100644 --- a/gateway/gateway-ujes-support/pom.xml +++ b/gateway/gateway-ujes-support/pom.xml @@ -51,6 +51,12 @@ <version>2.1</version> </dependency> + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cs-common</artifactId> + <version>${linkis.version}</version> + </dependency> + </dependencies> <build> diff --git a/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/parser/EntranceRequestGatewayParser.scala b/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/parser/EntranceRequestGatewayParser.scala index 348934e1c8..4775203889 100644 --- a/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/parser/EntranceRequestGatewayParser.scala +++ b/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/parser/EntranceRequestGatewayParser.scala @@ -19,12 +19,11 @@ package com.webank.wedatasphere.linkis.gateway.ujes.parser import com.webank.wedatasphere.linkis.common.ServiceInstance import com.webank.wedatasphere.linkis.gateway.http.GatewayContext import com.webank.wedatasphere.linkis.gateway.parser.AbstractGatewayParser +import com.webank.wedatasphere.linkis.gateway.ujes.parser.EntranceExecutionGatewayParser._ import com.webank.wedatasphere.linkis.protocol.constants.TaskConstant import com.webank.wedatasphere.linkis.protocol.utils.ZuulEntranceUtils import org.springframework.stereotype.Component -import EntranceExecutionGatewayParser._ - /** * created by cooperyang on 2019/5/15. */ diff --git a/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/route/HaContextGatewayRouter.scala b/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/route/HaContextGatewayRouter.scala new file mode 100644 index 0000000000..39ddbe9bfa --- /dev/null +++ b/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/route/HaContextGatewayRouter.scala @@ -0,0 +1,124 @@ +package com.webank.wedatasphere.linkis.gateway.ujes.route + +import java.util + +import com.webank.wedatasphere.linkis.common.ServiceInstance +import com.webank.wedatasphere.linkis.cs.common.entity.source.{ContextID, ContextIDParser} +import com.webank.wedatasphere.linkis.cs.common.protocol.ContextHTTPConstant +import com.webank.wedatasphere.linkis.cs.common.serialize.helper.ContextSerializationHelper +import com.webank.wedatasphere.linkis.gateway.http.GatewayContext +import com.webank.wedatasphere.linkis.gateway.route.AbstractGatewayRouter +import com.webank.wedatasphere.linkis.gateway.springcloud.SpringCloudGatewayConfiguration.{API_URL_PREFIX, normalPath} +import com.webank.wedatasphere.linkis.rpc.interceptor.ServiceInstanceUtils +import org.apache.commons.lang.StringUtils +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.stereotype.Component + +import scala.collection.JavaConversions._ +import scala.util.Random +import scala.util.matching.Regex + +/** + * created by cooperyang on 2020/2/19 + * Description: + */ +@Component +class HaContextGatewayRouter extends AbstractGatewayRouter{ + + @Autowired + private var contextIDParser: ContextIDParser = _ + private val serializationHelper = ContextSerializationHelper.getInstance(); + + override def route(gatewayContext: GatewayContext): ServiceInstance = { + + if (gatewayContext.getGatewayRoute.getRequestURI.contains("contextservice")){ + val params: util.HashMap[String, String] = gatewayContext.getGatewayRoute.getParams + if (!gatewayContext.getRequest.getQueryParams.isEmpty) { + for ((k, vArr) <- gatewayContext.getRequest.getQueryParams) { + if (vArr.nonEmpty) { + params.putIfAbsent(k, vArr.head) + } + } + } + if (gatewayContext.getRequest.getHeaders.containsKey(ContextHTTPConstant.CONTEXT_ID_STR)) { + params.putIfAbsent(ContextHTTPConstant.CONTEXT_ID_STR, gatewayContext.getRequest.getHeaders.get(ContextHTTPConstant.CONTEXT_ID_STR)(0)) + } + if (null == params || params.isEmpty) { + dealContextCreate(gatewayContext) + } else { + var contextId : String = null + for ((key, value) <- params) { + if (key.equalsIgnoreCase(ContextHTTPConstant.CONTEXT_ID_STR)) { + contextId = value + } + } + if (StringUtils.isNotBlank(contextId)) { + dealContextAccess(contextId.toString, gatewayContext) + } else { + dealContextCreate(gatewayContext) + } + } + }else{ + null + } + } + + def dealContextCreate(gatewayContext:GatewayContext):ServiceInstance = { + val serviceId = findService(HaContextGatewayRouter.CONTEXT_SERVICE_STR, list => { + val services = list.filter(_.contains(HaContextGatewayRouter.CONTEXT_SERVICE_STR)) + services.headOption + }) + val serviceInstances = ServiceInstanceUtils.getRPCServerLoader.getServiceInstances(serviceId.orNull) + if (serviceInstances.size > 0) { + val index = new Random().nextInt(serviceInstances.size) + serviceInstances(index) + } else { + logger.error(s"No valid instance for service : " + serviceId.orNull) + null + } + } + + def dealContextAccess(contextIdStr:String, gatewayContext: GatewayContext):ServiceInstance = { + val contextId : String = { + var tmpId : String = null + if (serializationHelper.accepts(contextIdStr)) { + val contextID : ContextID = serializationHelper.deserialize(contextIdStr).asInstanceOf[ContextID] + if (null != contextID) { + tmpId = contextID.getContextId + } else { + error(s"Deserializate contextID null. contextIDStr : " + contextIdStr) + } + } else { + error(s"ContxtIDStr cannot be deserialized. contextIDStr : " + contextIdStr) + } + if (null == tmpId) { + contextIdStr + } else { + tmpId + } + } + val instances = contextIDParser.parse(contextId) + var serviceId:Option[String] = None + serviceId = findService(HaContextGatewayRouter.CONTEXT_SERVICE_STR, list => { + val services = list.filter(_.contains(HaContextGatewayRouter.CONTEXT_SERVICE_STR)) + services.headOption + }) + val serviceInstances = ServiceInstanceUtils.getRPCServerLoader.getServiceInstances(serviceId.orNull) + if (instances.size() > 0) { + serviceId.map(ServiceInstance(_, instances.get(0))).orNull + } else if (serviceInstances.size > 0) { + serviceInstances(0) + } else { + logger.error(s"No valid instance for service : " + serviceId.orNull) + null + } + } + +} + + +object HaContextGatewayRouter{ + val CONTEXT_ID_STR:String = "contextId" + val CONTEXT_SERVICE_STR:String = "contextservice" + val CONTEXT_REGEX: Regex = (normalPath(API_URL_PREFIX) + "rest_[a-zA-Z][a-zA-Z_0-9]*/(v\\d+)/contextservice/" + ".+").r +} diff --git a/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/route/contextservice/ContextIdParserImpl.scala b/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/route/contextservice/ContextIdParserImpl.scala new file mode 100644 index 0000000000..a0fed9ea9c --- /dev/null +++ b/gateway/gateway-ujes-support/src/main/scala/com/webank/wedatasphere/linkis/gateway/ujes/route/contextservice/ContextIdParserImpl.scala @@ -0,0 +1,52 @@ +package com.webank.wedatasphere.linkis.gateway.ujes.route.contextservice + +import java.util +import java.util.{ArrayList, List} + +import com.webank.wedatasphere.linkis.common.utils.Logging +import com.webank.wedatasphere.linkis.cs.common.entity.source.{ContextIDParser, HAContextID} +import com.webank.wedatasphere.linkis.cs.common.utils.CSHighAvailableUtils +import com.webank.wedatasphere.linkis.rpc.instancealias.InstanceAliasManager +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.stereotype.Component + + +/** + * created by cooperyang on 2020/2/19 + * Description: 如果id为HAID，则解析出对应的instance + */ +@Component +class ContextIdParserImpl extends ContextIDParser with Logging { + + @Autowired + var instanceAliasManager : InstanceAliasManager = _ + + override def parse(contextId: String): util.List[String] = { + + if (CSHighAvailableUtils.checkHAIDBasicFormat(contextId)) { + val instances = new util.ArrayList[String](2) + val haContextID = CSHighAvailableUtils.decodeHAID(contextId) + if (instanceAliasManager.isInstanceAliasValid(haContextID.getInstance)) { + instances.add(instanceAliasManager.getInstanceByAlias(haContextID.getInstance).getInstance) + } else { + error(s"parse HAID instance invalid. haIDKey : " + contextId) + } + if (instanceAliasManager.isInstanceAliasValid(haContextID.getBackupInstance)) { + instances.add(instanceAliasManager.getInstanceByAlias(haContextID.getBackupInstance).getInstance) + } else { + error("parse HAID backupInstance invalid. haIDKey : " + contextId) + } + instances + } else { + new util.ArrayList[String](0) + } + } + + private def isNumberic(s:String):Boolean = { + s.toCharArray foreach { + c => if (c < 48 || c >57) return false + } + true + } + +} diff --git a/gateway/springcloudgateway/src/main/scala/com/webank/wedatasphere/linkis/gateway/springcloud/SpringCloudGatewayConfiguration.scala b/gateway/springcloudgateway/src/main/scala/com/webank/wedatasphere/linkis/gateway/springcloud/SpringCloudGatewayConfiguration.scala index 06b77a67a5..88b4cdf4b5 100644 --- a/gateway/springcloudgateway/src/main/scala/com/webank/wedatasphere/linkis/gateway/springcloud/SpringCloudGatewayConfiguration.scala +++ b/gateway/springcloudgateway/src/main/scala/com/webank/wedatasphere/linkis/gateway/springcloud/SpringCloudGatewayConfiguration.scala @@ -18,7 +18,6 @@ package com.webank.wedatasphere.linkis.gateway.springcloud import com.netflix.loadbalancer.Server import com.webank.wedatasphere.linkis.common.ServiceInstance -import com.webank.wedatasphere.linkis.common.conf.CommonVars import com.webank.wedatasphere.linkis.common.utils.Logging import com.webank.wedatasphere.linkis.gateway.config.GatewaySpringConfiguration import com.webank.wedatasphere.linkis.gateway.parser.{DefaultGatewayParser, GatewayParser} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
