Kaynağa Gözat

Configure python side of JT thrift to talk SASL

Todd Lipcon 15 yıl önce
ebeveyn
işleme
998be5ae3d

+ 1 - 2
desktop/libs/hadoop/src/hadoop/cluster.py

@@ -41,8 +41,7 @@ def _make_filesystem(identifier):
 
 def _make_mrcluster(identifier):
   cluster_conf = conf.MR_CLUSTERS[identifier]
-  return LiveJobTracker(cluster_conf.JT_HOST.get(),
-                        cluster_conf.JT_THRIFT_PORT.get())
+  return LiveJobTracker.from_conf(cluster_conf)
 
 FS_CACHE = None
 def get_hdfs(identifier="default"):

+ 7 - 2
desktop/libs/hadoop/src/hadoop/conf.py

@@ -115,7 +115,7 @@ HDFS_CLUSTERS = UnspecifiedConfigSection(
                             type=int),
       NN_HDFS_PORT=Config("hdfs_port", help="Hadoop IPC port for the name node", default=8020,
                             type=int),
-      NN_KERBEROS_PRINCIPAL=Config("kerberos_principal", help="Kerberos principal for NameNode",
+      NN_KERBEROS_PRINCIPAL=Config("nn_kerberos_principal", help="Kerberos principal for NameNode",
                                    default="hdfs", type=str),
       SECURITY_ENABLED=Config("security_enabled", help="Is running with Kerberos authentication",
                               default=False, type=bool),
@@ -131,7 +131,12 @@ MR_CLUSTERS = UnspecifiedConfigSection(
     members=dict(
       JT_HOST=Config("jobtracker_host", help="IP for JobTracker"),
       JT_THRIFT_PORT=Config("thrift_port", help="Thrift port for JobTracker", default=9290,
-                            type=int))))
+                            type=int),
+      JT_KERBEROS_PRINCIPAL=Config("jt_kerberos_principal", help="Kerberos principal for JobTracker",
+                                   default="mapred", type=str),
+      SECURITY_ENABLED=Config("security_enabled", help="Is running with Kerberos authentication",
+                              default=False, type=bool))
+))
 
 
 def config_validator():

+ 15 - 2
desktop/libs/hadoop/src/hadoop/job_tracker.py

@@ -46,7 +46,7 @@ def test_jt_configuration(cluster):
     return err
 
   try:
-    jt = LiveJobTracker(cluster.JT_HOST.get(), cluster.JT_THRIFT_PORT.get())
+    jt = LiveJobTracker.from_conf(cluster)
     jt.runtime_info()
   except TTransport.TTransportException:
     msg = 'Failed to contact JobTracker plugin at %s:%s.' % \
@@ -63,10 +63,14 @@ class LiveJobTracker(object):
   In particular, if Thrift returns None for anything, this will throw.
   """
 
-  def __init__(self, host, thrift_port):
+  def __init__(self, host, thrift_port,
+               security_enabled=False,
+               kerberos_principal="mapred"):
     self.client = thrift_util.get_client(
       Jobtracker.Client, host, thrift_port,
       service_name="Hadoop MR JobTracker HUE Plugin",
+      use_sasl=security_enabled,
+      kerberos_principal=kerberos_principal,
       timeout_seconds=JT_THRIFT_TIMEOUT)
     self.host = host
     self.thrift_port = thrift_port
@@ -76,6 +80,15 @@ class LiveJobTracker(object):
     self.thread_local = threading.local()
     self.setuser(DEFAULT_USER)
 
+  @classmethod
+  def from_conf(cls, conf):
+    return cls(
+      conf.JT_HOST.get(),
+      conf.JT_THRIFT_PORT.get(),
+      security_enabled=conf.SECURITY_ENABLED.get(),
+      kerberos_principal=conf.JT_KERBEROS_PRINCIPAL.get())
+
+
   def thriftjobid_from_string(self, jobid):
     """The jobid looks like this: job_201001301455_0001"""
     _, tid, jid = jobid.split("_")