Browse Source

HUE-8574 [importer] Setup automatically a Flume grapping Hue HTTPD logs and put into the sample collection

Romain Rigaux 7 years ago
parent
commit
2c01cd02ad

+ 3 - 1
apps/beeswax/src/beeswax/views.py

@@ -40,7 +40,7 @@ from desktop.lib.django_util import JsonResponse
 from desktop.lib.django_util import copy_query_dict, format_preserving_redirect, render
 from desktop.lib.django_util import login_notrequired, get_desktop_uri_prefix
 from desktop.lib.exceptions_renderable import PopupException
-from desktop.models import Document
+from desktop.models import Document, _get_apps
 from desktop.lib.parameterization import find_variables
 from desktop.views import serve_403_error
 from notebook.models import escape_rows
@@ -444,9 +444,11 @@ def execute_query(request, design_id=None, query_history_id=None):
     design = safe_get_design(request, query_type, design_id)
     query_history = None
 
+  current_app, other_apps, apps_list = _get_apps(request.user, '')
   doc = design and design.id and design.doc.get()
   context = {
     'design': design,
+    'apps': apps_list,
     'query': query_history, # Backward
     'query_history': query_history,
     'autocomplete_base_url': reverse(get_app_name(request) + ':api_autocomplete_databases', kwargs={}),

+ 0 - 0
desktop/libs/indexer/src/data/morphline/__init__.py


+ 114 - 0
desktop/libs/indexer/src/data/morphline/hue_accesslogs_no_geo.morphline.conf

@@ -0,0 +1,114 @@
+
+SOLR_LOCATOR : {
+    # Name of solr collection
+    collection : ${SOLR_COLLECTION}
+    # ZooKeeper ensemble
+    zkHost : "${ZOOKEEPER_ENSEMBLE}"
+}
+
+
+morphlines : [
+{
+    id : hue_accesslogs_no_geo
+
+    importCommands : ["org.kitesdk.**", "org.apache.solr.**"]
+    commands : [
+    {
+        ## Read the email stream and break it up into individual messages.
+        ## The beginning of a message is marked by regex clause below
+        ## The reason we use this command is that one event can have multiple
+        ## messages
+        readCSV {        
+        
+        ## Hue HTTPD load balancer
+        ## 172.18.18.3 - - [27/Aug/2018:05:47:12 -0700] "GET /static/desktop/js/jquery.rowselector.a04240f7cc48.js HTTP/1.1" 200 2321
+        
+      separator:  " "
+            columns:  [client_ip,C1,C2,time,dummy1,request,code,bytes]
+      ignoreFirstLine : false
+            quoteChar : "\""
+            commentPrefix : ""
+            trim : true
+            charset : UTF-8
+        }
+    }
+    {
+  split { 
+    inputField : request
+    outputFields : [method, url, protocol]          
+    separator : " "        
+    isRegex : false      
+    #separator : """\s*,\s*"""        
+    #  #isRegex : true      
+    addEmptyStrings : false
+    trim : true          
+          }
+    }
+     {
+  split { 
+    inputField : url 
+    outputFields : ["", app, subapp]          
+    separator : "\/"        
+    isRegex : false      
+    #separator : """\s*,\s*"""        
+    #  #isRegex : true      
+    addEmptyStrings : false
+    trim : true          
+          }
+    }
+    {
+  userAgent {
+    inputField : user_agent
+    outputFields : {
+      user_agent_family : "@{ua_family}"
+      user_agent_major  : "@{ua_major}"
+      device_family     : "@{device_family}"
+      os_family         : "@{os_family}"
+      os_major    : "@{os_major}"
+    }          
+  }
+    }
+
+      #{logInfo { format : "BODY : {}", args : ["@{}"] } }
+    # add Unique ID, in case our message_id field from above is not present
+    {
+        generateUUID {
+            field:id
+        }
+    }
+
+    # convert the timestamp field to "yyyy-MM-dd'T'HH:mm:ss.SSSZ" format
+    {
+       #  21/Nov/2014:22:08:27
+        convertTimestamp {
+            field : time 
+            inputFormats : ["[dd/MMM/yyyy:HH:mm:ss", "EEE, d MMM yyyy HH:mm:ss Z", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", "yyyy-MM-dd'T'HH:mm:ss", "yyyy-MM-dd"]
+            #inputTimezone : America/Los_Angeles
+            inputTimezone : UTC
+            outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
+            outputTimezone : UTC
+        }
+    }
+
+    # Consume the output record of the previous command and pipe another
+    # record downstream.
+    #
+    # This command sanitizes record fields that are unknown to Solr schema.xml
+    # by deleting them. Recall that Solr throws an exception on any attempt to
+    # load a document that contains a field that isn't specified in schema.xml
+    {
+        sanitizeUnknownSolrFields {
+            # Location from which to fetch Solr schema
+            solrLocator : ${SOLR_LOCATOR}
+        }
+    }
+
+    # load the record into a SolrServer or MapReduce SolrOutputFormat.
+    {
+        loadSolr {
+            solrLocator : ${SOLR_LOCATOR}
+        }
+    }
+    ]
+}
+]

+ 5 - 0
desktop/libs/indexer/src/indexer/conf.py

@@ -150,6 +150,11 @@ CONFIG_INDEXING_TEMPLATES_PATH = Config(
   default=os.path.join(os.path.dirname(__file__), '..', 'data', 'oozie_workspace')
   )
 
+
+def config_morphline_path():
+  return os.path.join(os.path.dirname(__file__), '..', 'data', 'morphline')
+
+
 # Unused
 SOLRCTL_PATH = Config(
   key="solrctl_path",

+ 26 - 11
desktop/libs/metadata/src/metadata/manager_api.py

@@ -18,8 +18,7 @@
 
 import json
 import logging
-
-from metadata.manager_client import ManagerApi
+import os
 
 from django.http import Http404
 from django.utils.html import escape
@@ -28,9 +27,12 @@ from django.views.decorators.http import require_POST
 
 from desktop.lib.django_util import JsonResponse
 from desktop.lib.i18n import force_unicode
+from libzookeeper.conf import zkensemble
+from indexer.conf import config_morphline_path
 
-from metadata.conf import has_navigator
 from metadata.catalog.navigator_client import CatalogApiException
+from metadata.conf import has_navigator
+from metadata.manager_client import ManagerApi
 
 
 LOG = logging.getLogger(__name__)
@@ -75,24 +77,37 @@ def hello(request):
 def update_flume_config(request):
   api = ManagerApi(request.user)
 
-  config = '''tier1.sources = source1
+  flume_agent_config = '''tier1.sources = source1
 tier1.channels = channel1
 tier1.sinks = sink1
- 
+
 tier1.sources.source1.type = exec
-tier1.sources.source1.command = tail -F /var/log/hue/access.log
+tier1.sources.source1.command = tail -F /var/log/hue-httpd/access_log
 tier1.sources.source1.channels = channel1
- 
+
 tier1.channels.channel1.type = memory
 tier1.channels.channel1.capacity = 10000
 tier1.channels.channel1.transactionCapacity = 1000
- 
+
 # Solr Sink configuration
 tier1.sinks.sink1.type          = org.apache.flume.sink.solr.morphline.MorphlineSolrSink
-tier1.sinks.sink1.morphlineFile = /tmp/morphline.conf
+tier1.sinks.sink1.morphlineFile = morphlines.conf
+tier1.sinks.sink1.morphlineId = hue_accesslogs_no_geo
 tier1.sinks.sink1.channel       = channel1'''
 
-  response = api.update_and_refresh_flume(cluster_name=None, config=config)
 
-  return JsonResponse(response)
+  morphline_config = open(os.path.join(config_morphline_path(), 'hue_accesslogs_no_geo.morphline.conf')).read()
+  morphline_config = morphline_config.replace(
+    '${SOLR_COLLECTION}', 'log_analytics_demo'
+  ).replace(
+    '${ZOOKEEPER_ENSEMBLE}', '%s/solr' % zkensemble()
+  )
+
+  responses = {}
+
+  responses['agent_config_file'] = api.update_flume_config(cluster_name=None, config_name='agent_config_file', config_value=flume_agent_config)
+  responses['agent_morphlines_conf_file'] = api.update_flume_config(cluster_name=None, config_name='agent_morphlines_conf_file', config_value=morphline_config)
+
+  responses['refresh_flume'] = api.refresh_flume(cluster_name=None, restart=True)
 
+  return JsonResponse(responses)

+ 23 - 5
desktop/libs/metadata/src/metadata/manager_client.py

@@ -134,7 +134,7 @@ class ManagerApi(object):
       raise ManagerApiException(e)
 
 
-  def update_flume_config(self, cluster_name, config):
+  def update_flume_config(self, cluster_name, config_name, config_value):
     service = 'FLUME-1'
     cluster = self._get_cluster(cluster_name)
     roleConfigGroup = [role['roleConfigGroupRef']['roleConfigGroupName'] for role in self._get_roles(cluster['name'], service, 'AGENT')]
@@ -143,7 +143,7 @@ class ManagerApi(object):
         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[0]),
         u'body': {
           u'items': [
-            {u'name': u'agent_config_file', u'value': config}
+            {u'name': config_name, u'value': config_value}
           ]
         },
         u'contentType': u'application/json',
@@ -156,13 +156,15 @@ class ManagerApi(object):
     )
 
 
-  def update_and_refresh_flume(self, cluster_name, config):
+  def refresh_flume(self, cluster_name, restart=False):
     service = 'FLUME-1'
     cluster = self._get_cluster(cluster_name)
     roles = [role['name'] for role in self._get_roles(cluster['name'], service, 'AGENT')]
 
-    self.update_flume_config(cluster['name'], config)
-    return self.refresh_configs(cluster['name'], service, roles)
+    if restart:    
+      return self.restart_services(cluster['name'], service, roles)
+    else:
+      return self.refresh_configs(cluster['name'], service, roles)
 
 
   def refresh_configs(self, cluster_name, service=None, roles=None):
@@ -181,6 +183,22 @@ class ManagerApi(object):
       raise ManagerApiException(e)
 
 
+  def restart_services(self, cluster_name, service=None, roles=None):
+    try:
+      if service is None:
+        return self._root.post('clusters/%(cluster_name)s/commands/restart' % {'cluster_name': cluster_name}, contenttype="application/json")
+      elif roles is None:
+        return self._root.post('clusters/%(cluster_name)s/services/%(service)s/roleCommands/restart' % {'cluster_name': cluster_name, 'service': service}, contenttype="application/json")
+      else:
+        return self._root.post(
+            'clusters/%(cluster_name)s/services/%(service)s/roleCommands/restart' % {'cluster_name': cluster_name, 'service': service},
+            data=json.dumps({"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')