You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@kyuubi.apache.org by GitBox <gi...@apache.org> on 2022/06/04 11:47:01 UTC
[GitHub] [incubator-kyuubi] mahmoudbahaa opened a new pull request, #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
mahmoudbahaa opened a new pull request, #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815
<!--
Thanks for sending a pull request!
Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->
### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
1. If you add a feature, you can talk about the use case of it.
2. If you fix a bug, you can clarify why it is a bug.
-->
Add http transport protocol support to Kyuubi
main code was taken from:
1. https://github.com/SteNicholas/incubator-kyuubi/commit/25ac0731d7331616988026996ca918793af151e9
2. https://github.com/apache/hive/blob/branch-3.1/service/src/java/org/apache/hive/service/cli/thrift/ThriftHttpCLIService.java
2. https://github.com/apache/hive/blob/branch-3.1/service/src/java/org/apache/hive/service/cli/thrift/ThriftHttpServlet.java
### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [ ] Add screenshots for manual tests if appropriate
Tested against NONE and LDAP,
Not tested against KERBEROS or CUSTOM
- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891435441
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.{THRIFT_CONN_FAIL, THRIFT_CONN_OPEN, THRIFT_CONN_TOTAL}
+import org.apache.kyuubi.metrics.MetricsSystem
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection, please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection. please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.doc)
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("Thrift HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("Thrift HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
+ connector.setReuseAddress(!OSUtils.IS_WINDOWS)
+ val maxIdleTime = conf.get(FRONTEND_THRIFT_HTTP_MAX_IDLE_TIME)
+ connector.setIdleTimeout(maxIdleTime)
+ connector.setAcceptQueueSize(maxThreads)
+ server.foreach(_.addConnector(connector))
+
+ val processor = new TCLIService.Processor[TCLIService.Iface](this)
+ val protocolFactory = new TBinaryProtocol.Factory
+ val servlet = new ThriftHttpServlet(processor, protocolFactory, authFactory, conf)
+ servlet.init()
+
+ // Context handler
+ val context = new ServletContextHandler(ServletContextHandler.SESSIONS)
+ context.setContextPath("/")
+
+ context.addEventListener(new ServletContextListener() {
+ override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.incCount(THRIFT_CONN_TOTAL)
+ ms.incCount(THRIFT_CONN_OPEN)
Review Comment:
we should inc/dec it in thrift binary as well or just change the name
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891371801
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.{THRIFT_CONN_FAIL, THRIFT_CONN_OPEN, THRIFT_CONN_TOTAL}
+import org.apache.kyuubi.metrics.MetricsSystem
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection, please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection. please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.doc)
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("Thrift HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("Thrift HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
+ connector.setReuseAddress(!OSUtils.IS_WINDOWS)
+ val maxIdleTime = conf.get(FRONTEND_THRIFT_HTTP_MAX_IDLE_TIME)
+ connector.setIdleTimeout(maxIdleTime)
+ connector.setAcceptQueueSize(maxThreads)
+ server.foreach(_.addConnector(connector))
+
+ val processor = new TCLIService.Processor[TCLIService.Iface](this)
+ val protocolFactory = new TBinaryProtocol.Factory
+ val servlet = new ThriftHttpServlet(processor, protocolFactory, authFactory, conf)
+ servlet.init()
+
+ // Context handler
+ val context = new ServletContextHandler(ServletContextHandler.SESSIONS)
+ context.setContextPath("/")
+
+ context.addEventListener(new ServletContextListener() {
+ override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.incCount(THRIFT_CONN_TOTAL)
+ ms.incCount(THRIFT_CONN_OPEN)
+ }
+ }
+
+ override def contextDestroyed(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.decCount(THRIFT_CONN_OPEN)
+ }
+ }
+ })
+
+ val httpPath = getHttpPath(conf.get(FRONTEND_THRIFT_HTTP_PATH))
+
+ if (conf.get(FRONTEND_THRIFT_HTTP_COMPRESSION_ENABLED)) {
+ val gzipHandler = new GzipHandler
+ gzipHandler.setHandler(context)
+ gzipHandler.addIncludedMethods(HttpMethod.POST.asString())
+ gzipHandler.addIncludedMimeTypes(APPLICATION_THRIFT)
+ server.foreach(_.setHandler(gzipHandler))
+ } else {
+ server.foreach(_.setHandler(context))
+ }
+
+ context.addServlet(new ServletHolder(servlet), httpPath)
+ constrainHttpMethods(context)
+
+ info("Started " + getClass.getSimpleName +
+ " in " + schemeName + " mode on port " + portNum +
+ " path=" + httpPath + " with " + minThreads + "..." + maxThreads + " worker threads")
+ } catch {
+ case e: Throwable =>
+ MetricsSystem.tracing(_.incCount(THRIFT_CONN_FAIL))
+ error(e)
Review Comment:
redundant?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] ulysses-you commented on pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
ulysses-you commented on PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#issuecomment-1146957835
thank you @mahmoudbahaa working on this ! cc @turboFei @yanghua @yaooqinn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891385106
##########
dev/reformat:
##########
@@ -22,4 +22,4 @@ KYUUBI_HOME="$(cd "`dirname "$0"`/.."; pwd)"
PROFILES="-Pflink-provided,hive-provided,spark-provided,spark-block-cleaner,spark-3.3,spark-3.2,spark-3.1,spark-3.0,tpcds"
-${KYUUBI_HOME}/build/mvn spotless:apply $PROFILES
+mvn spotless:apply $PROFILES
Review Comment:
Please revert this change when this PR get ready, the `${KYUUBI_HOME}/build/mvn` will pick up the local `mvn` if there is apposite `mvn` in your PATH rather than download a new one
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891381605
##########
.gitignore:
##########
@@ -74,3 +74,6 @@ embedded_zookeeper/
**/operation_logs/
**/server_operation_logs/
**/engine_operation_logs/
+conf/log4j2.properties
+conf/kyuubi-defaults.conf
+conf/kyuubi-env.sh
Review Comment:
I think it's a good idea to exclude those files then we can debug with specific configurations in local easily
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891377550
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.{THRIFT_CONN_FAIL, THRIFT_CONN_OPEN, THRIFT_CONN_TOTAL}
+import org.apache.kyuubi.metrics.MetricsSystem
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection, please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection. please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.doc)
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("Thrift HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("Thrift HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
+ connector.setReuseAddress(!OSUtils.IS_WINDOWS)
+ val maxIdleTime = conf.get(FRONTEND_THRIFT_HTTP_MAX_IDLE_TIME)
+ connector.setIdleTimeout(maxIdleTime)
+ connector.setAcceptQueueSize(maxThreads)
+ server.foreach(_.addConnector(connector))
+
+ val processor = new TCLIService.Processor[TCLIService.Iface](this)
+ val protocolFactory = new TBinaryProtocol.Factory
+ val servlet = new ThriftHttpServlet(processor, protocolFactory, authFactory, conf)
+ servlet.init()
+
+ // Context handler
+ val context = new ServletContextHandler(ServletContextHandler.SESSIONS)
+ context.setContextPath("/")
+
+ context.addEventListener(new ServletContextListener() {
+ override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.incCount(THRIFT_CONN_TOTAL)
+ ms.incCount(THRIFT_CONN_OPEN)
+ }
+ }
+
+ override def contextDestroyed(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.decCount(THRIFT_CONN_OPEN)
+ }
+ }
+ })
+
+ val httpPath = getHttpPath(conf.get(FRONTEND_THRIFT_HTTP_PATH))
+
+ if (conf.get(FRONTEND_THRIFT_HTTP_COMPRESSION_ENABLED)) {
+ val gzipHandler = new GzipHandler
+ gzipHandler.setHandler(context)
+ gzipHandler.addIncludedMethods(HttpMethod.POST.asString())
+ gzipHandler.addIncludedMimeTypes(APPLICATION_THRIFT)
+ server.foreach(_.setHandler(gzipHandler))
+ } else {
+ server.foreach(_.setHandler(context))
+ }
+
+ context.addServlet(new ServletHolder(servlet), httpPath)
+ constrainHttpMethods(context)
+
+ info("Started " + getClass.getSimpleName +
+ " in " + schemeName + " mode on port " + portNum +
+ " path=" + httpPath + " with " + minThreads + "..." + maxThreads + " worker threads")
+ } catch {
+ case e: Throwable =>
+ MetricsSystem.tracing(_.incCount(THRIFT_CONN_FAIL))
+ error(e)
+ throw new KyuubiException(
+ s"Failed to initialize frontend service on $serverAddr:$portNum.",
+ e)
+ }
+ super.initialize(conf)
+ }
+
+ override def run(): Unit =
+ try {
+ if (isServer()) {
+ info(s"Starting and exposing JDBC connection at: jdbc:hive2://$connectionUrl/")
+ }
+ server.foreach(_.start())
+ } catch {
+ case _: InterruptedException => error(s"$getName is interrupted")
+ case t: Throwable =>
+ error(s"Error starting $getName", t)
+ System.exit(-1)
+ }
+
+ override protected def stopServer(): Unit = {
+ server.foreach(_.stop())
+ server = None
+ }
+
+ override protected def isServer(): Boolean = true
+
+ override val discoveryService: Option[Service] = None
+
+ private def getHttpPath(httpPath: String): String = {
+ if (httpPath == null || httpPath == "") return "/*"
+ else {
+ if (!httpPath.startsWith("/")) return "/" + httpPath
+ if (httpPath.endsWith("/")) return httpPath + "*"
+ if (!httpPath.endsWith("/*")) return httpPath + "/*"
+ }
+ httpPath
+ }
+
+ def constrainHttpMethods(ctxHandler: ServletContextHandler): Unit = {
+ val constraint = new Constraint
+ constraint.setAuthenticate(true)
+ val cmt = new ConstraintMapping
+ cmt.setConstraint(constraint)
+ cmt.setMethod("TRACE")
+ cmt.setPathSpec("/*")
+ val securityHandler = new ConstraintSecurityHandler
+ val cmo = new ConstraintMapping
+ cmo.setConstraint(constraint)
+ cmo.setMethod("OPTIONS")
+ cmo.setPathSpec("/*")
+ securityHandler.setConstraintMappings(Array[ConstraintMapping](cmt, cmo))
+ ctxHandler.setSecurityHandler(securityHandler)
+ }
+
+ override protected def getIpAddress: String = {
+ SessionManager.getIpAddress
+ }
+
+ override protected def getUserName(req: TOpenSessionReq): String = {
+ var userName: String = SessionManager.getUserName
+ if (userName == null) userName = req.getUsername
+ userName = getShortName(userName)
+ val effectiveClientUser: String = getProxyUser(req.getConfiguration, getIpAddress, userName)
+ debug("Client's username: " + effectiveClientUser)
+ effectiveClientUser
+ }
+
+ private def getShortName(userName: String): String = {
+ var ret: String = null
+
+ if (userName != null) {
+ if (authFactory.isKerberosEnabled) {
+ // KerberosName.getShorName can only be used for kerberos user, but not for the user
+ // logged in via other authentications such as LDAP
+ val fullKerberosName = ShimLoader.getHadoopShims.getKerberosNameShim(userName)
+ ret = fullKerberosName.getShortName
Review Comment:
question: kyuubi support enable kerberos and LDAP in same time, is it sufficient to handle that case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891435441
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.{THRIFT_CONN_FAIL, THRIFT_CONN_OPEN, THRIFT_CONN_TOTAL}
+import org.apache.kyuubi.metrics.MetricsSystem
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection, please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection. please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.doc)
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("Thrift HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("Thrift HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
+ connector.setReuseAddress(!OSUtils.IS_WINDOWS)
+ val maxIdleTime = conf.get(FRONTEND_THRIFT_HTTP_MAX_IDLE_TIME)
+ connector.setIdleTimeout(maxIdleTime)
+ connector.setAcceptQueueSize(maxThreads)
+ server.foreach(_.addConnector(connector))
+
+ val processor = new TCLIService.Processor[TCLIService.Iface](this)
+ val protocolFactory = new TBinaryProtocol.Factory
+ val servlet = new ThriftHttpServlet(processor, protocolFactory, authFactory, conf)
+ servlet.init()
+
+ // Context handler
+ val context = new ServletContextHandler(ServletContextHandler.SESSIONS)
+ context.setContextPath("/")
+
+ context.addEventListener(new ServletContextListener() {
+ override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.incCount(THRIFT_CONN_TOTAL)
+ ms.incCount(THRIFT_CONN_OPEN)
Review Comment:
we should inc/dec it in thrift binary as well or just change the name to match what's it really represent
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891373271
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.{THRIFT_CONN_FAIL, THRIFT_CONN_OPEN, THRIFT_CONN_TOTAL}
+import org.apache.kyuubi.metrics.MetricsSystem
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection, please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection. please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.doc)
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("Thrift HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("Thrift HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
+ connector.setReuseAddress(!OSUtils.IS_WINDOWS)
+ val maxIdleTime = conf.get(FRONTEND_THRIFT_HTTP_MAX_IDLE_TIME)
+ connector.setIdleTimeout(maxIdleTime)
+ connector.setAcceptQueueSize(maxThreads)
+ server.foreach(_.addConnector(connector))
+
+ val processor = new TCLIService.Processor[TCLIService.Iface](this)
+ val protocolFactory = new TBinaryProtocol.Factory
+ val servlet = new ThriftHttpServlet(processor, protocolFactory, authFactory, conf)
+ servlet.init()
+
+ // Context handler
+ val context = new ServletContextHandler(ServletContextHandler.SESSIONS)
+ context.setContextPath("/")
+
+ context.addEventListener(new ServletContextListener() {
+ override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.incCount(THRIFT_CONN_TOTAL)
+ ms.incCount(THRIFT_CONN_OPEN)
+ }
+ }
+
+ override def contextDestroyed(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.decCount(THRIFT_CONN_OPEN)
+ }
+ }
+ })
+
+ val httpPath = getHttpPath(conf.get(FRONTEND_THRIFT_HTTP_PATH))
+
+ if (conf.get(FRONTEND_THRIFT_HTTP_COMPRESSION_ENABLED)) {
+ val gzipHandler = new GzipHandler
+ gzipHandler.setHandler(context)
+ gzipHandler.addIncludedMethods(HttpMethod.POST.asString())
+ gzipHandler.addIncludedMimeTypes(APPLICATION_THRIFT)
+ server.foreach(_.setHandler(gzipHandler))
+ } else {
+ server.foreach(_.setHandler(context))
+ }
+
+ context.addServlet(new ServletHolder(servlet), httpPath)
+ constrainHttpMethods(context)
+
+ info("Started " + getClass.getSimpleName +
+ " in " + schemeName + " mode on port " + portNum +
+ " path=" + httpPath + " with " + minThreads + "..." + maxThreads + " worker threads")
+ } catch {
+ case e: Throwable =>
+ MetricsSystem.tracing(_.incCount(THRIFT_CONN_FAIL))
+ error(e)
+ throw new KyuubiException(
+ s"Failed to initialize frontend service on $serverAddr:$portNum.",
+ e)
+ }
+ super.initialize(conf)
+ }
+
+ override def run(): Unit =
+ try {
+ if (isServer()) {
+ info(s"Starting and exposing JDBC connection at: jdbc:hive2://$connectionUrl/")
Review Comment:
Beeline default use thrift binary mode, we need another parameters for thrift http mode
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] codecov-commenter commented on pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#issuecomment-1146655164
# [Codecov](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#2815](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (46e7e9c) into [master](https://codecov.io/gh/apache/incubator-kyuubi/commit/b42c28b5ac42d4456fb8e133935fc21a7370bd32?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b42c28b) will **decrease** coverage by `0.61%`.
> The diff coverage is `43.77%`.
```diff
@@ Coverage Diff @@
## master #2815 +/- ##
============================================
- Coverage 62.84% 62.22% -0.62%
Complexity 202 202
============================================
Files 480 486 +6
Lines 21339 21956 +617
Branches 2833 2901 +68
============================================
+ Hits 13411 13663 +252
- Misses 6651 6999 +348
- Partials 1277 1294 +17
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/kyuubi/server/http/util/HttpAuthUtils.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2ZXIvaHR0cC91dGlsL0h0dHBBdXRoVXRpbHMuc2NhbGE=) | `0.00% <0.00%> (ø)` | |
| [...pache/kyuubi/server/http/util/SessionManager.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2ZXIvaHR0cC91dGlsL1Nlc3Npb25NYW5hZ2VyLnNjYWxh) | `0.00% <0.00%> (ø)` | |
| [...kyuubi/server/statestore/api/SessionMetadata.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2ZXIvc3RhdGVzdG9yZS9hcGkvU2Vzc2lvbk1ldGFkYXRhLnNjYWxh) | `95.23% <0.00%> (ø)` | |
| [...g/apache/kyuubi/session/KyuubiSessionManager.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXNzaW9uL0t5dXViaVNlc3Npb25NYW5hZ2VyLnNjYWxh) | `85.36% <ø> (-1.22%)` | :arrow_down: |
| [.../apache/kyuubi/server/http/util/CookieSigner.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2ZXIvaHR0cC91dGlsL0Nvb2tpZVNpZ25lci5zY2FsYQ==) | `6.89% <6.89%> (ø)` | |
| [...e/authentication/KyuubiAuthenticationFactory.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLWNvbW1vbi9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2aWNlL2F1dGhlbnRpY2F0aW9uL0t5dXViaUF1dGhlbnRpY2F0aW9uRmFjdG9yeS5zY2FsYQ==) | `72.05% <10.00%> (-10.70%)` | :arrow_down: |
| [.../apache/kyuubi/server/http/ThriftHttpServlet.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2ZXIvaHR0cC9UaHJpZnRIdHRwU2VydmxldC5zY2FsYQ==) | `12.26% <12.26%> (ø)` | |
| [...a/org/apache/kyuubi/service/TFrontendService.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLWNvbW1vbi9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2aWNlL1RGcm9udGVuZFNlcnZpY2Uuc2NhbGE=) | `94.65% <40.00%> (-0.30%)` | :arrow_down: |
| [...apache/kyuubi/service/TBinaryFrontendService.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLWNvbW1vbi9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2aWNlL1RCaW5hcnlGcm9udGVuZFNlcnZpY2Uuc2NhbGE=) | `87.75% <50.00%> (-1.61%)` | :arrow_down: |
| [...che/kyuubi/server/KyuubiTHttpFrontendService.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2ZXIvS3l1dWJpVEh0dHBGcm9udGVuZFNlcnZpY2Uuc2NhbGE=) | `51.09% <51.09%> (ø)` | |
| ... and [24 more](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [9265cf3...46e7e9c](https://codecov.io/gh/apache/incubator-kyuubi/pull/2815?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891440687
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/http/ThriftHttpServlet.scala:
##########
@@ -0,0 +1,539 @@
+/*
+ * 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.kyuubi.server.http
+
+import java.io.IOException
+import java.security.PrivilegedExceptionAction
+import java.security.SecureRandom
+import javax.security.sasl.AuthenticationException
+import javax.servlet.ServletException
+import javax.servlet.http.Cookie
+import javax.servlet.http.HttpServletRequest
+import javax.servlet.http.HttpServletResponse
+import javax.ws.rs.core.NewCookie
+
+import scala.collection.mutable
+
+import org.apache.commons.codec.binary.Base64
+import org.apache.commons.codec.binary.StringUtils
+import org.apache.hadoop.hive.shims.HadoopShims.KerberosNameShim
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.hive.shims.Utils
+import org.apache.hadoop.security.UserGroupInformation
+import org.apache.thrift.TProcessor
+import org.apache.thrift.protocol.TProtocolFactory
+import org.apache.thrift.server.TServlet
+import org.ietf.jgss.GSSContext
+import org.ietf.jgss.GSSCredential
+import org.ietf.jgss.GSSException
+import org.ietf.jgss.GSSManager
+import org.ietf.jgss.Oid
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.server.http.authentication.KerberosAuthenticationHandler
+import org.apache.kyuubi.server.http.util.{CookieSigner, HttpAuthUtils, SessionManager}
+import org.apache.kyuubi.service.authentication.{AuthenticationProviderFactory, KyuubiAuthenticationFactory}
+
+class ThriftHttpServlet(
+ processor: TProcessor,
+ protocolFactory: TProtocolFactory,
+ authFactory: KyuubiAuthenticationFactory,
+ conf: KyuubiConf)
+ extends TServlet(processor, protocolFactory) with Logging {
+ // Class members for cookie based authentication.
+ private var signer: CookieSigner = _
+ val AUTH_COOKIE = "hive.server2.auth"
+ private val RAN = new SecureRandom()
+ private var isCookieAuthEnabled: Boolean = false
+ private var cookieDomain: String = _
+ private var cookiePath: String = _
+ private var cookieMaxAge: Int = 0
+ private var isCookieSecure = false
+ private var isHttpOnlyCookie = false
+ private val HIVE_DELEGATION_TOKEN_HEADER = "X-Hive-Delegation-Token"
+ private val X_FORWARDED_FOR = "X-Forwarded-For"
Review Comment:
Please unify the name style of `HIVE_DELEGATION_TOKEN_HEADER` and `X_FORWARDED_FOR`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] mahmoudbahaa commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
mahmoudbahaa commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r892755073
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.{THRIFT_CONN_FAIL, THRIFT_CONN_OPEN, THRIFT_CONN_TOTAL}
+import org.apache.kyuubi.metrics.MetricsSystem
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection, please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection. please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.doc)
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("Thrift HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("Thrift HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
+ connector.setReuseAddress(!OSUtils.IS_WINDOWS)
+ val maxIdleTime = conf.get(FRONTEND_THRIFT_HTTP_MAX_IDLE_TIME)
+ connector.setIdleTimeout(maxIdleTime)
+ connector.setAcceptQueueSize(maxThreads)
+ server.foreach(_.addConnector(connector))
+
+ val processor = new TCLIService.Processor[TCLIService.Iface](this)
+ val protocolFactory = new TBinaryProtocol.Factory
+ val servlet = new ThriftHttpServlet(processor, protocolFactory, authFactory, conf)
+ servlet.init()
+
+ // Context handler
+ val context = new ServletContextHandler(ServletContextHandler.SESSIONS)
+ context.setContextPath("/")
+
+ context.addEventListener(new ServletContextListener() {
+ override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.incCount(THRIFT_CONN_TOTAL)
+ ms.incCount(THRIFT_CONN_OPEN)
+ }
+ }
+
+ override def contextDestroyed(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.decCount(THRIFT_CONN_OPEN)
+ }
+ }
+ })
+
+ val httpPath = getHttpPath(conf.get(FRONTEND_THRIFT_HTTP_PATH))
+
+ if (conf.get(FRONTEND_THRIFT_HTTP_COMPRESSION_ENABLED)) {
+ val gzipHandler = new GzipHandler
+ gzipHandler.setHandler(context)
+ gzipHandler.addIncludedMethods(HttpMethod.POST.asString())
+ gzipHandler.addIncludedMimeTypes(APPLICATION_THRIFT)
+ server.foreach(_.setHandler(gzipHandler))
+ } else {
+ server.foreach(_.setHandler(context))
+ }
+
+ context.addServlet(new ServletHolder(servlet), httpPath)
+ constrainHttpMethods(context)
+
+ info("Started " + getClass.getSimpleName +
+ " in " + schemeName + " mode on port " + portNum +
+ " path=" + httpPath + " with " + minThreads + "..." + maxThreads + " worker threads")
+ } catch {
+ case e: Throwable =>
+ MetricsSystem.tracing(_.incCount(THRIFT_CONN_FAIL))
+ error(e)
+ throw new KyuubiException(
+ s"Failed to initialize frontend service on $serverAddr:$portNum.",
+ e)
+ }
+ super.initialize(conf)
+ }
+
+ override def run(): Unit =
+ try {
+ if (isServer()) {
+ info(s"Starting and exposing JDBC connection at: jdbc:hive2://$connectionUrl/")
Review Comment:
hhtp mode is enabled in beeline/jdbc using the `transportMode=http` option for example `jdbc:hive2://localhost:10010/default;transportMode=http`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] mahmoudbahaa closed pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
mahmoudbahaa closed pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
URL: https://github.com/apache/incubator-kyuubi/pull/2815
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#issuecomment-1148840926
Thanks @mahmoudbahaa, I take a quick look on this PR and left a few comments, overall it's in good shape, looking forward the UT and some uncompeleted methods. You can split this PR into some small ones if you think this PR is too large.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yanghua commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
yanghua commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r889792873
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,285 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.common.metrics.common.{MetricsConstant, MetricsFactory}
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection")
Review Comment:
Can we point out the missed information detail of the config item?
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,285 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.common.metrics.common.{MetricsConstant, MetricsFactory}
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection")
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection")
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
+ connector.setReuseAddress(!OSUtils.IS_WINDOWS)
+ val maxIdleTime = conf.get(FRONTEND_THRIFT_HTTP_MAX_IDLE_TIME)
+ connector.setIdleTimeout(maxIdleTime)
+ connector.setAcceptQueueSize(maxThreads)
+ server.foreach(_.addConnector(connector))
+
+ val processor = new TCLIService.Processor[TCLIService.Iface](this)
+ val protocolFactory = new TBinaryProtocol.Factory
+ val servlet = new ThriftHttpServlet(processor, protocolFactory, authFactory, conf)
+ servlet.init()
+
+ // Context handler
+ val context = new ServletContextHandler(ServletContextHandler.SESSIONS)
+ context.setContextPath("/")
+
+ context.addEventListener(new ServletContextListener() {
+ override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = {
+ val metrics = MetricsFactory.getInstance
+ if (metrics != null) {
+ try {
+ metrics.incrementCounter(MetricsConstant.OPEN_CONNECTIONS)
+ metrics.incrementCounter(MetricsConstant.CUMULATIVE_CONNECTION_COUNT)
Review Comment:
Glad to see we report some metrics here. Can we also make sure if it is suitable for Kyuubi's metric system?
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,285 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.common.metrics.common.{MetricsConstant, MetricsFactory}
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection")
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection")
Review Comment:
ditto
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,285 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.common.metrics.common.{MetricsConstant, MetricsFactory}
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection")
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection")
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("HTTP Server SSL: adding excluded protocols: " +
Review Comment:
More readable log information? Kyuubi has owned an HTTP server to serve restful API. So let's make the `HTTP server` clear?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891371362
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.{THRIFT_CONN_FAIL, THRIFT_CONN_OPEN, THRIFT_CONN_TOTAL}
+import org.apache.kyuubi.metrics.MetricsSystem
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection, please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection. please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.doc)
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("Thrift HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("Thrift HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
+ connector.setReuseAddress(!OSUtils.IS_WINDOWS)
+ val maxIdleTime = conf.get(FRONTEND_THRIFT_HTTP_MAX_IDLE_TIME)
+ connector.setIdleTimeout(maxIdleTime)
+ connector.setAcceptQueueSize(maxThreads)
+ server.foreach(_.addConnector(connector))
+
+ val processor = new TCLIService.Processor[TCLIService.Iface](this)
+ val protocolFactory = new TBinaryProtocol.Factory
+ val servlet = new ThriftHttpServlet(processor, protocolFactory, authFactory, conf)
+ servlet.init()
+
+ // Context handler
+ val context = new ServletContextHandler(ServletContextHandler.SESSIONS)
+ context.setContextPath("/")
+
+ context.addEventListener(new ServletContextListener() {
+ override def contextInitialized(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.incCount(THRIFT_CONN_TOTAL)
+ ms.incCount(THRIFT_CONN_OPEN)
+ }
+ }
+
+ override def contextDestroyed(servletContextEvent: ServletContextEvent): Unit = {
+ MetricsSystem.tracing { ms =>
+ ms.decCount(THRIFT_CONN_OPEN)
+ }
+ }
+ })
+
+ val httpPath = getHttpPath(conf.get(FRONTEND_THRIFT_HTTP_PATH))
+
+ if (conf.get(FRONTEND_THRIFT_HTTP_COMPRESSION_ENABLED)) {
+ val gzipHandler = new GzipHandler
+ gzipHandler.setHandler(context)
+ gzipHandler.addIncludedMethods(HttpMethod.POST.asString())
+ gzipHandler.addIncludedMimeTypes(APPLICATION_THRIFT)
+ server.foreach(_.setHandler(gzipHandler))
+ } else {
+ server.foreach(_.setHandler(context))
+ }
+
+ context.addServlet(new ServletHolder(servlet), httpPath)
+ constrainHttpMethods(context)
+
+ info("Started " + getClass.getSimpleName +
+ " in " + schemeName + " mode on port " + portNum +
+ " path=" + httpPath + " with " + minThreads + "..." + maxThreads + " worker threads")
Review Comment:
let's use s"text $var1 ${var2.var3}"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r891369255
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.kyuubi.server
+
+import java.net.ServerSocket
+import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
+import javax.servlet.{ServletContextEvent, ServletContextListener}
+
+import org.apache.hadoop.hive.shims.ShimLoader
+import org.apache.hadoop.shaded.org.jline.utils.OSUtils
+import org.apache.hive.service.rpc.thrift.{TCLIService, TOpenSessionReq}
+import org.apache.thrift.protocol.TBinaryProtocol
+import org.eclipse.jetty.http.HttpMethod
+import org.eclipse.jetty.security.{ConstraintMapping, ConstraintSecurityHandler}
+import org.eclipse.jetty.server._
+import org.eclipse.jetty.server.handler.gzip.GzipHandler
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.security.Constraint
+import org.eclipse.jetty.util.ssl.SslContextFactory
+import org.eclipse.jetty.util.thread.ExecutorThreadPool
+
+import org.apache.kyuubi.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants.{THRIFT_CONN_FAIL, THRIFT_CONN_OPEN, THRIFT_CONN_TOTAL}
+import org.apache.kyuubi.metrics.MetricsSystem
+import org.apache.kyuubi.server.http.ThriftHttpServlet
+import org.apache.kyuubi.server.http.util.SessionManager
+import org.apache.kyuubi.service.{Serverable, Service, ServiceUtils, TFrontendService}
+import org.apache.kyuubi.util.NamedThreadFactory
+
+/**
+ * Apache Thrift based hive service rpc
+ * 1. the server side implementation serves client-server rpc calls
+ * 2. the engine side implementations serve server-engine rpc calls
+ */
+final class KyuubiTHttpFrontendService(
+ override val serverable: Serverable)
+ extends TFrontendService("KyuubiTHttpFrontendService") {
+
+ override protected lazy val serverHost: Option[String] =
+ conf.get(FRONTEND_THRIFT_HTTP_BIND_HOST)
+ override protected lazy val portNum: Int = conf.get(FRONTEND_THRIFT_HTTP_BIND_PORT)
+ override protected lazy val actualPort: Int = portNum
+ override protected lazy val serverSocket: ServerSocket = null
+
+ private var server: Option[Server] = None
+ private val APPLICATION_THRIFT = "application/x-thrift"
+
+ /**
+ * Configure Jetty to serve http requests. Example of a client connection URL:
+ * http://localhost:10000/servlets/thrifths2/ A gateway may cause actual target
+ * URL to differ, e.g. http://gateway:port/hive2/servlets/thrifths2/.
+ *
+ * @param conf the configuration of the service
+ */
+ override def initialize(conf: KyuubiConf): Unit = synchronized {
+ this.conf = conf
+ try {
+ // Server thread pool
+ // Start with minWorkerThreads, expand till maxWorkerThreads and reject
+ // subsequent requests
+ val minThreads = conf.get(FRONTEND_THRIFT_MIN_WORKER_THREADS)
+ val maxThreads = conf.get(FRONTEND_THRIFT_MAX_WORKER_THREADS)
+ val keepAliveTime = conf.get(FRONTEND_THRIFT_WORKER_KEEPALIVE_TIME)
+ val executor = new ThreadPoolExecutor(
+ minThreads,
+ maxThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue[Runnable](),
+ new NamedThreadFactory(getName + "HttpHandler-Pool", false))
+ val threadPool = new ExecutorThreadPool(executor)
+
+ // HTTP Server
+ server = Some(new Server(threadPool))
+
+ val httpConf = new HttpConfiguration
+ // Configure header size
+ val requestHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE)
+ val responseHeaderSize = conf.get(FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE)
+ httpConf.setRequestHeaderSize(requestHeaderSize)
+ httpConf.setResponseHeaderSize(responseHeaderSize)
+ val connectionFactory = new HttpConnectionFactory(httpConf)
+
+ val useSsl = conf.get(FRONTEND_THRIFT_HTTP_USE_SSL)
+ val schemeName = if (useSsl) "https" else "http"
+
+ // Change connector if SSL is used
+ val connector =
+ if (useSsl) {
+ val keyStorePath = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH)
+
+ if (keyStorePath.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.key +
+ " Not configured for SSL connection, please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PATH.doc)
+ }
+
+ val keyStorePassword = conf.get(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD)
+ if (keyStorePassword.isEmpty) {
+ throw new IllegalArgumentException(FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.key +
+ " Not configured for SSL connection. please set the key with: " +
+ FRONTEND_THRIFT_HTTP_SSL_KEYSTORE_PASSWORD.doc)
+ }
+
+ val sslContextFactory = new SslContextFactory.Server
+ val excludedProtocols = conf.get(FRONTEND_THRIFT_HTTP_SSL_PROTOCOL_BLACKLIST).split(",")
+ info("Thrift HTTP Server SSL: adding excluded protocols: " +
+ String.join(",", excludedProtocols: _*))
+ sslContextFactory.addExcludeProtocols(excludedProtocols: _*)
+ info("Thrift HTTP Server SSL: SslContextFactory.getExcludeProtocols = " +
+ String.join(",", sslContextFactory.getExcludeProtocols: _*))
+ sslContextFactory.setKeyStorePath(keyStorePath.get)
+ sslContextFactory.setKeyStorePassword(keyStorePassword.get)
+ new ServerConnector(
+ server.get,
+ sslContextFactory,
+ connectionFactory)
+ } else {
+ new ServerConnector(server.get, connectionFactory)
+ }
+
+ connector.setPort(portNum)
+ // Linux:yes, Windows:no
Review Comment:
maybe Linux => Unix-like ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#discussion_r890701418
##########
kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala:
##########
@@ -434,6 +452,121 @@ object KyuubiConf {
.version("1.4.0")
.fallbackConf(FRONTEND_LOGIN_BACKOFF_SLOT_LENGTH)
+ val FRONTEND_THRIFT_HTTP_REQUEST_HEADER_SIZE: ConfigEntry[Int] =
+ buildConf("kyuubi.frontend.thrift.http.request.header.size")
+ .doc("Request header size in bytes, when using HTTP transport mode. Jetty defaults used.")
+ .version("1.6.0")
+ .intConf
+ .createWithDefault(6 * 1024)
+
+ val FRONTEND_THRIFT_HTTP_RESPONSE_HEADER_SIZE: ConfigEntry[Int] =
+ buildConf("frontend.thrift.http.response.header.size")
Review Comment:
"frontend.xxx" => "kyuubi.frontend.xxx"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] mahmoudbahaa commented on pull request #2815: [KYUUBI #886] Add HTTP transport mode support to KYUUBI
Posted by GitBox <gi...@apache.org>.
mahmoudbahaa commented on PR #2815:
URL: https://github.com/apache/incubator-kyuubi/pull/2815#issuecomment-1150283647
I made a new PR with no kerberos support for now (can be added at later date) as it is not yet feature complete nor tested so I will be closing this PR and we can continue discussion on the other PR https://github.com/apache/incubator-kyuubi/pull/2833
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org