Explorar o código

HUE-8551 [importer] Support setting basic Flume configs

Romain Rigaux %!s(int64=7) %!d(string=hai) anos
pai
achega
d52f6aa

+ 1 - 1
desktop/core/src/desktop/templates/ko_components/ko_multi_cluster_sidebar.mako

@@ -169,7 +169,7 @@ from desktop.views import _ko
       }, {
         label: '${ _('Storage') }',
         items: [{
-            label: '${ _('SDX Catalog') }',
+            label: '${ _('Catalog') }',
             url: '/',
             icon: 'altus-icon altus-sdx'
           },{

+ 9 - 0
desktop/libs/indexer/src/indexer/api3.py

@@ -519,6 +519,15 @@ def _envelope_job(request, file_format, destination, start_time=None, lib_path=N
         "kafkaFieldTypes": file_format['kafkaFieldTypes']
       }
 
+      if True:
+        properties['window'] = ''
+      else: # For "KafkaSQL"
+        properties['window'] = '''
+            window {
+                enabled = true
+                milliseconds = 60000
+            }'''
+
     if destination['outputFormat'] == 'table':
       if destination['isTargetExisting']:
         # Todo: check if format matches

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

@@ -109,7 +109,7 @@ SPARK_KAFKA_VERSION=0.10 spark2-submit envelope.jar envelope.conf"""
       if properties['streamSelection'] == 'kafka':
         input = """type = kafka
                 brokers = "%(brokers)s"
-                topic = %(topics)s
+                topics = [%(topics)s]
                 encoding = string
                 translator {
                     type = %(kafkaFieldType)s
@@ -117,10 +117,7 @@ SPARK_KAFKA_VERSION=0.10 spark2-submit envelope.jar envelope.conf"""
                     field.names = [%(kafkaFieldNames)s]
                     field.types = [%(kafkaFieldTypes)s]
                 }
-                window {
-                    enabled = true
-                    milliseconds = 60000
-                }
+                %(window)s
         """ % properties
       elif properties['streamSelection'] == 'sfdc':
         input = """type = sfdc
@@ -210,7 +207,7 @@ SPARK_KAFKA_VERSION=0.10 spark2-submit envelope.jar envelope.conf"""
     return """
 application {
     name = %(app_name)s
-    %(batch)
+    %(batch)s
     executors = 1
     executor.cores = 1
     executor.memory = 1G

+ 54 - 0
desktop/libs/metadata/src/metadata/manager_client.py

@@ -16,7 +16,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import logging
+import urllib
 
 from django.core.cache import cache
 from django.utils.translation import ugettext as _
@@ -132,6 +134,58 @@ class ManagerApi(object):
       raise ManagerApiException(e)
 
 
+  def update_flume_config(self, cluster_name, config):
+    service = 'FLUME-1'
+    roleConfigGroup = [role['roleConfigGroupRef']['roleConfigGroupName'] for role in self._get_roles(cluster_name, service, 'AGENT')]
+    data = {
+      u'items': [{
+        u'url': u'/api/v8/clusters/%(cluster_name)s/services/%(service)s/roleConfigGroups/%(roleConfigGroups)s/config?message=Updated%20service%20and%20role%20type%20configurations.'.replace('%(cluster_name)s', urllib.quote(cluster_name)).replace('%(service)s', service).replace('%(roleConfigGroups)s', roleConfigGroup),
+        u'body': {
+          u'items': [
+            {u'name': u'agent_config_file', u'value': config}
+          ]
+        },
+        u'contentType': u'application/json',
+        u'method': u'PUT'
+      }]
+    }
+
+    return self.batch(
+      items=data
+    )
+
+
+  def update_and_refresh_flume(self, cluster_name, config):
+    service = 'FLUME-1'
+    roles = [role['name'] for role in self._get_roles(cluster_name, service, 'AGENT')]
+
+    self.update_flume_config(cluster_name, config)
+    self.refresh_configs(cluster_name, service, roles)
+
+
+  def refresh_configs(self, cluster_name, service=None, roles=None):
+    try:
+      if service is None:
+        return self._root.post('clusters/%(cluster_name)s/commands/refresh' % {'cluster_name': cluster_name}, contenttype="application/json")
+      elif roles is None:
+        return self._root.post('clusters/%(cluster_name)s/services/%(service)s/commands/refresh' % {'cluster_name': cluster_name, 'service': service}, contenttype="application/json")
+      else:
+        return self._root.post(
+            'clusters/%(cluster_name)s/services/%(service)s/commands/refresh' % {'cluster_name': cluster_name, 'service': service},
+            data=json.dump({"items": roles}),
+            contenttype="application/json"
+        )
+    except RestException, e:
+      raise ManagerApiException(e)
+
+
+  def batch(self, items):
+    try:
+      return self._root.post('batch', data=json.dumps(items), contenttype='application/json')
+    except RestException, e:
+      raise ManagerApiException(e)
+
+
   def _get_cluster(self, cluster_name=None):
     clusters = self._root.get('clusters/')['items']