Quellcode durchsuchen

HUE-1957 [core] Configure thrift transport

Allow framed and buffered for now.
Abraham Elmahrek vor 11 Jahren
Ursprung
Commit
0520f6c

+ 2 - 1
apps/hbase/src/hbase/api.py

@@ -90,7 +90,8 @@ class HbaseApi(object):
                                   service_name="Hue HBase Thrift Client for %s" % name,
                                   kerberos_principal=None,
                                   use_sasl=False,
-                                  timeout_seconds=None)
+                                  timeout_seconds=None,
+                                  transport=conf.THRIFT_TRANSPORT.get())
 
   def get(self, cluster, tableName, row, column, attributes):
     client = self.connectCluster(cluster)

+ 18 - 1
apps/hbase/src/hbase/conf.py

@@ -16,7 +16,7 @@
 # limitations under the License.
 
 
-from desktop.lib.conf import Config
+from desktop.lib.conf import Config, validate_thrift_transport
 
 HBASE_CLUSTERS = Config(
   key="hbase_clusters",
@@ -29,3 +29,20 @@ TRUNCATE_LIMIT = Config(
   default="500",
   help="Hard limit of rows or columns per row fetched before truncating.",
   type=int)
+
+THRIFT_TRANSPORT = Config(
+  key="thrift_transport",
+  default="buffered",
+  help="'buffered' is the default of the HBase Thrift Server. " +
+       "'framed' can be used to chunk up responses, " +
+       "which is useful when used in conjunction with the nonblocking server in Thrift.",
+  type=str
+)
+
+
+def config_validator(user):
+  res = []
+
+  res.extend(validate_thrift_transport(THRIFT_TRANSPORT))
+
+  return res

+ 5 - 0
desktop/conf.dist/hue.ini

@@ -800,6 +800,11 @@
   # Hard limit of rows or columns per row fetched before truncating.
   ## truncate_limit = 500
 
+  # 'buffered' is the default of the HBase Thrift Server.
+  # 'framed' can be used to chunk up responses,
+  # which is useful when used in conjunction with the nonblocking server in Thrift.
+  ## thrift_transport=buffered
+
 
 ###########################################################################
 # Settings to configure Solr Search

+ 5 - 0
desktop/conf/pseudo-distributed.ini.tmpl

@@ -841,6 +841,11 @@
   # Hard limit of rows or columns per row fetched before truncating.
   ## truncate_limit = 500
 
+  # 'buffered' is the default of the HBase Thrift Server.
+  # 'framed' can be used to chunk up responses,
+  # which is useful when used in conjunction with the nonblocking server in Thrift.
+  ## thrift_transport=buffered
+
 
 ###########################################################################
 # Settings to configure Solr Search

+ 16 - 0
desktop/core/src/desktop/lib/conf.py

@@ -80,6 +80,9 @@ import sys
 # Magical object for use as a "symbol"
 _ANONYMOUS = ("_ANONYMOUS")
 
+# Supported thrift transports
+SUPPORTED_THRIFT_TRANSPORTS = ('buffered', 'framed')
+
 # a BoundContainer(BoundConfig) object which has all of the application's configs as members
 GLOBAL_CONFIG = None
 
@@ -676,3 +679,16 @@ def validate_port(confvar):
   except ValueError:
     return error_res
   return [ ]
+
+def validate_thrift_transport(confvar):
+  """
+  Validate that the provided thrift transport is supported.
+  Returns [(confvar, error_msg)] or []
+  """
+  transport = confvar.get()
+  error_res = [(confvar, 'Thrift transport %s not supported. Please choose a supported transport: %s' % (transport, ', '.join(SUPPORTED_THRIFT_TRANSPORTS)))]
+
+  if transport not in SUPPORTED_THRIFT_TRANSPORTS:
+    return error_res
+
+  return []

+ 10 - 3
desktop/core/src/desktop/lib/thrift_util.py

@@ -28,7 +28,7 @@ import sys
 from thrift.Thrift import TType, TApplicationException
 from thrift.transport.TSocket import TSocket
 from thrift.transport.TSSLSocket import TSSLSocket
-from thrift.transport.TTransport import TBufferedTransport, TMemoryBuffer,\
+from thrift.transport.TTransport import TBufferedTransport, TFramedTransport, TMemoryBuffer,\
                                         TTransportException
 from thrift.protocol.TBinaryProtocol import TBinaryProtocol
 from desktop.lib.python_util import create_synchronous_io_multiplexer
@@ -78,7 +78,8 @@ class ConnectionConfig(object):
                keyfile=None,
                certfile=None,
                validate=False,
-               timeout_seconds=45):
+               timeout_seconds=45,
+               transport='buffered'):
     """
     @param klass The thrift client class
     @param host Host to connect to
@@ -96,6 +97,7 @@ class ConnectionConfig(object):
     @param certfile certificate file
     @param validate Validate the certificate received from server
     @param timeout_seconds Timeout for thrift calls
+    @param transport string representation of thrift transport to use
     """
     self.klass = klass
     self.host = host
@@ -111,10 +113,11 @@ class ConnectionConfig(object):
     self.certfile = certfile
     self.validate = validate
     self.timeout_seconds = timeout_seconds
+    self.transport = transport
 
   def __str__(self):
     return ', '.join(map(str, [self.klass, self.host, self.port, self.service_name, self.use_sasl, self.kerberos_principal, self.timeout_seconds,
-                               self.mechanism, self.username, self.use_ssl, self.ca_certs, self.keyfile, self.certfile, self.validate]))
+                               self.mechanism, self.username, self.use_ssl, self.ca_certs, self.keyfile, self.certfile, self.validate, self.transport]))
 
 class ConnectionPooler(object):
   """
@@ -248,6 +251,8 @@ def connect_to_thrift(conf):
       return saslc
 
     transport = TSaslClientTransport(sasl_factory, conf.mechanism, sock)
+  elif conf.transport == 'framed':
+    transport = TFramedTransport(sock)
   else:
     transport = TBufferedTransport(sock)
 
@@ -267,6 +272,8 @@ def _grab_transport_from_wrapper(outer_transport):
     return outer_transport._TBufferedTransport__trans
   elif isinstance(outer_transport, TSaslClientTransport):
     return outer_transport._trans
+  elif isinstance(outer_transport, TFramedTransport):
+    return outer_transport._TFramedTransport__trans
   else:
     raise Exception("Unknown transport type: " + outer_transport.__class__)