Browse Source

HUE-8298 [metadata] Get Kafka brokers hosts via Manager API

Romain Rigaux 7 years ago
parent
commit
32fd0f3cab

+ 3 - 2
desktop/libs/indexer/src/indexer/indexers/envelope.py

@@ -80,7 +80,8 @@ class EnvelopeIndexer(object):
     properties = {
       "brokers": "self-service-analytics-1.gce.cloudera.com:9092,self-service-analytics-2.gce.cloudera.com:9092,self-service-analytics-3.gce.cloudera.com:9092",
       "kudu_master": "self-service-analytics-1.gce.cloudera.com:7051",
-      "output_table": "impala::default.traffic_conditions"
+      "output_table": "impala::default.traffic_conditions",
+      "topics": "traffic"
     }
 
     return """
@@ -97,7 +98,7 @@ steps {
         input {
             type = kafka
             brokers = "%(brokers)s"
-            topics = traffic
+            topics = %(topics)s
             encoding = string
             translator {
                 type = delimited

+ 1 - 12
desktop/libs/metadata/src/metadata/manager_api.py

@@ -32,7 +32,7 @@ from django.utils.translation import ugettext as _
 from django.views.decorators.http import require_POST
 
 from desktop.lib.django_util import JsonResponse
-from desktop.lib.i18n import force_unicode, smart_unicode
+from desktop.lib.i18n import force_unicode
 
 from metadata.conf import has_navigator
 from metadata.navigator_client import NavigatorApiException
@@ -41,17 +41,6 @@ from metadata.navigator_client import NavigatorApiException
 LOG = logging.getLogger(__name__)
 
 
-class ManagerApiException(Exception):
-  def __init__(self, message=None):
-    self.message = message or _('No error message, please check the logs.')
-
-  def __str__(self):
-    return str(self.message)
-
-  def __unicode__(self):
-    return smart_unicode(self.message)
-
-
 def error_handler(view_fn):
   def decorator(*args, **kwargs):
     status = 500

+ 36 - 4
desktop/libs/metadata/src/metadata/manager_client.py

@@ -22,16 +22,26 @@ from django.core.cache import cache
 from django.utils.translation import ugettext as _
 
 from desktop.lib.rest.http_client import RestException, HttpClient
-
-from metadata.conf import MANAGER
 from desktop.lib.rest.resource import Resource
-from metadata.manager_api import ManagerApiException
+from desktop.lib.i18n import smart_unicode
+from metadata.conf import MANAGER
 
 
 LOG = logging.getLogger(__name__)
 VERSION = 'v19'
 
 
+class ManagerApiException(Exception):
+  def __init__(self, message=None):
+    self.message = message or _('No error message, please check the logs.')
+
+  def __str__(self):
+    return str(self.message)
+
+  def __unicode__(self):
+    return smart_unicode(self.message)
+
+
 class ManagerApi(object):
   """
   https://cloudera.github.io/cm_api/
@@ -63,5 +73,27 @@ class ManagerApi(object):
       LOG.info(params)
       return self._root.get('tools/echo', params=params)
     except RestException, e:
-      LOG.error('Failed to search for entities with search query')
+      raise ManagerApiException(e)
+
+
+  def get_kafka_brokers(self, cluster_name=None):
+    try:
+      clusters = self._root.get('clusters/')['items']
+
+      if len(clusters) > 1:
+        cluster = [cluster for cluster in clusters if cluster['name'] == cluster_name]
+      else:
+        cluster = clusters[0]
+
+      services = self._root.get('clusters/%(name)s/services' % cluster)['items']
+      kafka_service = [service for service in services if service['type'] == 'KAFKA'][0]
+
+      kafka_roles = self._root.get('clusters/%(name)s/services/%(kafka_service)s/roles' % {'name': cluster['name'], 'kafka_service': kafka_service['name']})['items']
+      kafka_broker_hostids = [broker_hostid['hostRef']['hostId'] for broker_hostid in kafka_roles if  broker_hostid['type'] == 'KAFKA_BROKER']
+
+      hosts = self._root.get('hosts')['items']
+      kafka_brokers_hosts = [host['hostname'] + ':9092' for host in hosts if host['hostId'] in kafka_broker_hostids]
+
+      return ','.join(kafka_brokers_hosts)
+    except RestException, e:
       raise ManagerApiException(e)