Browse Source

HUE-4449 [indexer] Add ruby log to indexer

Aaron Peddle 9 năm trước cách đây
mục cha
commit
76d944f

+ 21 - 2
desktop/libs/indexer/src/data/oozie_workspace/clean_to_match_schema.conf

@@ -75,8 +75,27 @@
         ]
         then : []
         else : [
-          { logInfo { format : "Ignoring record due to invalid date for ${field['name']}: {}", args : ["@{}"] } }
-          { dropRecord {} }
+        # the built in convertTimestamp options failed, try some custom ones
+          {
+            if {
+              conditions : [
+                {
+                  convertTimestamp {
+                    field : date
+                    inputFormats : ["dd/MMM/yyyy HH:mm:ss Z", "dd/MMM/yyyy:HH:mm:ss Z"]
+                    outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
+                    outputTimezone : UTC
+                  }
+                }
+              ]
+              then : []
+              else : [
+                # none of the custom dates matched either. Drop the record
+                { logInfo { format : "Ignoring record due to invalid date for ${field['name']}: {}", args : ["@{}"] } }
+                { dropRecord {} }
+              ]
+            }
+          }
         ]
       }
     }

+ 1 - 1
desktop/libs/indexer/src/data/oozie_workspace/morphline_template.conf

@@ -42,7 +42,7 @@ morphlines : [
         }
       }
 
-      <%include file="${'parse_%s.conf' % format['type']}", args="format=format, format_character=format_character"/>
+      <%include file="${'parse_%s.conf' % format_class.get_parse_type()}", args="format_settings=format_settings, format_class=format_class"/>
 
 
       {

+ 0 - 60
desktop/libs/indexer/src/data/oozie_workspace/parse_combined_apache.conf

@@ -1,60 +0,0 @@
-# Licensed to Cloudera, Inc. under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  Cloudera, Inc. licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.from nose.tools import assert_equal
-<%page args="format, format_character"/>
-
-{
-  readLine{
-    charset: UTF-8
-  }
-}
-
-
-{
-  tryRules {
-    catchExceptions : true
-    throwExceptionIfAllRulesFailed : false
-    rules : [
-      {
-        commands : [
-          {
-            grok {
-              dictionaryFiles : ["grok_dictionaries"]
-
-              expressions : {
-                message: """%{COMBINEDAPACHELOG}"""
-              }
-            }
-          }
-        ]
-      }
-      {
-        # do nothing if match of grok failed
-        commands : [
-          { logInfo { format : "failed to grok line {}", args : ["@{}"] } }
-        ]
-      }
-    ]
-  }
-}
-
-{
-  convertTimestamp {
-    field : timestamp
-    inputFormats : ["dd/MMM/yyyy:HH:mm:ss Z"]
-    outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
-    outputTimezone : UTC
-  }
-}

+ 4 - 4
desktop/libs/indexer/src/data/oozie_workspace/parse_csv.conf

@@ -13,18 +13,18 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.from nose.tools import assert_equal
-<%page args="format, format_character"/>
+<%page args="format_settings, format_class"/>
 
 {
   readCSV {
-    separator : "${format_character(format['fieldSeparator'])}"
+    separator : "${format_class.format_character(format_settings['fieldSeparator'])}"
     columns : [
       % for field in fields[:num_base_fields]:
       "${field['name']}"
       %endfor
     ]
-    quoteChar : "${format_character(format['quoteChar'])}"
-    ignoreFirstLine : "${'true' if format['hasHeader'] else 'false'}"
+    quoteChar : "${format_class.format_character(format_settings['quoteChar'])}"
+    ignoreFirstLine : "${'true' if format_settings['hasHeader'] else 'false'}"
     charset : UTF-8
   }
 }

+ 67 - 0
desktop/libs/indexer/src/data/oozie_workspace/parse_grok_line.conf

@@ -0,0 +1,67 @@
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.from nose.tools import assert_equal
+<%page args="format_settings, format_class"/>
+
+{
+  readLine{
+    charset: UTF-8
+  }
+}
+
+# rename field from message to file_line since message is ocassionally used by grok commands as an output field
+{
+  addValues {
+    file_line : "@{message}"
+  }
+}
+{
+  removeFields {
+    blacklist : ["literal:message"]
+  }
+}
+
+{
+  if{
+    conditions : [
+      # check if the grok will work and if it does then do it. Actions in conditions don't persist
+      {
+        grok {
+          dictionaryFiles : ["grok_dictionaries"]
+
+          expressions : {
+            input_line: """${format_class.get_grok()}"""
+          }
+        }
+      }
+    ]
+    then : [
+      {
+        grok {
+          dictionaryFiles : ["grok_dictionaries"]
+
+          expressions : {
+            input_line: """${format_class.get_grok()}"""
+          }
+        }
+      }
+      { logInfo { format : "Successfuly grokked {}", args : ["@{}"] } }
+    ]
+    else : [
+      { logInfo { format : "Ignoring record due to ungrokable format: {}", args : ["@{}"] } }
+      { dropRecord {} }
+    ]
+  }
+}

+ 1 - 10
desktop/libs/indexer/src/data/oozie_workspace/parse_hue.conf

@@ -13,7 +13,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.from nose.tools import assert_equal
-<%page args="format, format_character"/>
+<%page args="format_instance, format_class"/>
 
 {
   readMultiLine{
@@ -62,12 +62,3 @@
   }
 }
 { logTrace { format : "Output from multiline grok {}", args : ["@{}"] } }
-
-{
-  convertTimestamp {
-    field : date
-    inputFormats : ["dd/MMM/yyyy HH:mm:ss Z"]
-    outputFormat : "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
-    outputTimezone : UTC
-  }
-}

+ 132 - 35
desktop/libs/indexer/src/indexer/file_format.py

@@ -29,29 +29,34 @@ def get_format_types():
   return [
     CSVFormat,
     HueFormat,
-    ApacheFormat
+    ApacheCombinedFormat,
+    RubyLog
   ]
 
-def get_format_mapping():
+def _get_format_mapping():
   return dict([(format_.get_name(), format_) for format_ in get_format_types()])
 
+def get_file_format_class(type_):
+  mapping = _get_format_mapping()
+  return mapping[type_] if type_ in mapping else None
+
 def get_file_format_instance(file, format_=None):
   file_stream = file['stream']
   file_extension = file['name'].split('.')[-1] if '.' in file['name'] else ''
 
-  format_mapping = get_format_mapping()
+  format_mapping = _get_format_mapping()
 
   if format_ and "type" in format_:
     type_ = format_["type"]
     if type_ in format_mapping:
       if format_mapping[type_].valid_format(format_):
-        return format_mapping[type_](file_stream, format_)
+        return format_mapping[type_].get_instance(file_stream, format_)
       else:
         return None
 
   matches = [type_ for type_ in get_format_types() if file_extension in type_.get_extensions()]
 
-  return (matches[0] if matches else get_format_types()[0])(file_stream, format_)
+  return (matches[0] if matches else get_format_types()[0]).get_instance(file_stream, format_)
 
 class FileFormat(object):
   _name = None
@@ -59,6 +64,7 @@ class FileFormat(object):
   _customizable = True
   _args = []
   _extensions = []
+  _parse_type = None
 
   @classmethod
   def get_extensions(cls):
@@ -68,6 +74,10 @@ class FileFormat(object):
   def get_name(cls):
     return cls._name
 
+  @classmethod
+  def get_parse_type(cls):
+    return cls._parse_type if cls._parse_type else cls.get_name()
+
   @classmethod
   def get_description(cls):
     return cls._description
@@ -90,9 +100,14 @@ class FileFormat(object):
       "name": cls.get_name(),
       "args": [arg.to_dict() for arg in cls.get_arguments()],
       "description": cls.get_description(),
-      "isCustomizable": cls.is_customizable()
+      "isCustomizable": cls.is_customizable(),
+      "parse_type": cls.get_parse_type()
     }
 
+  @classmethod
+  def get_instance(cls, file_stream, format_):
+    return cls()
+
   def __init__(self):
     pass
 
@@ -105,7 +120,7 @@ class FileFormat(object):
     return []
 
   def get_format(self):
-    return {"type": self.get_name()}
+    return {"type": self.get_name(), "parse_type": self.get_parse_type()}
 
   def get_fields(self):
     obj = {}
@@ -124,13 +139,29 @@ class FileFormat(object):
 
     return obj
 
-class HueFormat(FileFormat):
+class GrokkedFormat(FileFormat):
+  _grok = None
+
+  @classmethod
+  def get_grok(cls):
+    return cls._grok
+
+  def get_format(self):
+    format_ = super(GrokkedFormat, self).get_format()
+    specific_format = {
+      "grok":self.get_grok()
+    }
+    format_.update(specific_format)
+
+    return format_
+
+class HueFormat(GrokkedFormat):
   _name = "hue"
   _description = _("Hue Log File")
   _customizable = False
   _extensions = ["log"]
 
-  def __init__(self, file_stream, format_):
+  def __init__(self):
     self._fields = [
       Field("date", "date"),
       Field("component", "string"),
@@ -148,13 +179,17 @@ class HueFormat(FileFormat):
   def fields(self):
     return self._fields
 
-class ApacheFormat(FileFormat):
+class GrokLineFormat(GrokkedFormat):
+  _parse_type = "grok_line"
+
+class ApacheCombinedFormat(GrokLineFormat):
   _name = "combined_apache"
   _description = _("Combined Apache Log File")
   _customizable = False
   _extensions = ["log"]
+  _grok = "%{COMBINEDAPACHELOG}"
 
-  def __init__(self, file_stream, format_):
+  def __init__(self):
     self._fields = [
       Field("clientip", "string"),
       Field("ident", "string"),
@@ -167,7 +202,28 @@ class ApacheFormat(FileFormat):
       Field("response", "long"),
       Field("bytes", "long"),
       Field("referrer", "string"),
-      Field("message", "text_en")
+      Field("field_line", "text_en")
+    ]
+
+  @property
+  def fields(self):
+    return self._fields
+
+class RubyLog(GrokLineFormat):
+  _name = "ruby_log"
+  _description = _("Ruby Log")
+  _customizable = False
+  _extensions = ["log"]
+  _grok = "%{RUBY_LOGGER}"
+
+  def __init__(self):
+    self._fields = [
+      Field("timestamp", "string"),
+      Field("pid", "long"),
+      Field("loglevel", "string"),
+      Field("progname", "string"),
+      Field("message", "text_en"),
+      Field("field_line", "text_en")
     ]
 
   @property
@@ -185,10 +241,26 @@ class CSVFormat(FileFormat):
   ]
   _extensions = ["csv", "tsv"]
 
+  @staticmethod
+  def format_character(string):
+    string = string.replace('\\', '\\\\')
+    string = string.replace('"', '\\"')
+    string = string.replace('\t', '\\t')
+    string = string.replace('\n', '\\n')
+
+    return string
+
   @classmethod
   def _valid_character(self, char):
     return isinstance(char, basestring) and len(char) == 1
 
+  @classmethod
+  def _guess_dialect(cls, sample):
+    sniffer = csv.Sniffer()
+    dialect = sniffer.sniff(sample)
+    has_header = sniffer.has_header(sample)
+    return dialect, has_header
+
   @classmethod
   def valid_format(cls, format_):
     valid = super(CSVFormat, cls).valid_format(format_)
@@ -199,28 +271,59 @@ class CSVFormat(FileFormat):
 
     return valid
 
-  def __init__(self, file_stream, format_=None):
+  @classmethod
+  def _guess_from_file_stream(cls, file_stream):
+    file_stream.seek(0)
+    sample = '\n'.join(file_stream.read(1024*1024*5).splitlines())
+    file_stream.seek(0)
+
+    try:
+      dialect, has_header = cls._guess_dialect(sample)
+      delimiter = dialect.delimiter
+      line_terminator = dialect.lineterminator
+      quote_char = dialect.quotechar
+    except Exception:
+      # guess dialect failed, fall back to defaults:
+      return cls()
+
+    return cls(**{
+      "delimiter":delimiter,
+      "line_terminator": line_terminator,
+      "quote_char": quote_char,
+      "has_header": has_header,
+      "sample": sample
+    })
+
+  @classmethod
+  def _from_format(cls, file_stream, format_):
     file_stream.seek(0)
     sample = '\n'.join(file_stream.read(1024*1024*5).splitlines())
     file_stream.seek(0)
 
-    if self.valid_format(format_):
-      self._delimiter = format_["fieldSeparator"].encode('utf-8')
-      self._line_terminator = format_["recordSeparator"].encode('utf-8')
-      self._quote_char = format_["quoteChar"].encode('utf-8')
-      self._has_header = format_["hasHeader"]
+    delimiter = format_["fieldSeparator"].encode('utf-8')
+    line_terminator = format_["recordSeparator"].encode('utf-8')
+    quote_char = format_["quoteChar"].encode('utf-8')
+    has_header = format_["hasHeader"]
+    return cls(**{
+      "delimiter":delimiter,
+      "line_terminator": line_terminator,
+      "quote_char": quote_char,
+      "has_header": has_header,
+      "sample": sample
+    })
+
+  @classmethod
+  def get_instance(cls, file_stream, format_):
+    if cls.valid_format(format_):
+      return cls._from_format(file_stream, format_)
     else:
-      try:
-        dialect, self._has_header = self._guess_dialect(sample)
-        self._delimiter = dialect.delimiter
-        self._line_terminator = dialect.lineterminator
-        self._quote_char = dialect.quotechar
-      except Exception:
-        # guess dialect failed, fall back to defaults:
-        self._delimiter = ','
-        self._line_terminator = '\n'
-        self._quote_char = '"'
-        self._has_header = False
+      return cls._guess_from_file_stream(file_stream)
+
+  def __init__(self, delimiter=',', line_terminator='\n', quote_char='"', has_header=False, sample=""):
+    self._delimiter = delimiter
+    self._line_terminator = line_terminator
+    self._quote_char = quote_char
+    self._has_header = has_header
 
     # sniffer insists on \r\n even when \n. This is safer and good enough for a preview
     self._line_terminator = self._line_terminator.replace("\r\n", "\n")
@@ -265,12 +368,6 @@ class CSVFormat(FileFormat):
 
     return format_
 
-  def _guess_dialect(self, sample):
-    sniffer = csv.Sniffer()
-    dialect = sniffer.sniff(sample)
-    has_header = sniffer.has_header(sample)
-    return dialect, has_header
-
   def _guess_num_columns(self, sample_rows):
     counts = {}
 

+ 3 - 12
desktop/libs/indexer/src/indexer/smart_indexer.py

@@ -25,7 +25,7 @@ from liboozie.submission2 import Submission
 
 from indexer.fields import Field, FIELD_TYPES
 from indexer.operations import get_checked_args
-from indexer.file_format import get_file_format_instance
+from indexer.file_format import get_file_format_instance, get_file_format_class
 from indexer.conf import CONFIG_INDEXING_TEMPLATES_PATH
 from indexer.conf import CONFIG_INDEXER_LIBS_PATH
 from indexer.conf import zkensemble
@@ -143,15 +143,6 @@ class Indexer(object):
 
     return base_name
 
-  @staticmethod
-  def _format_character(string):
-    string = string.replace('\\', '\\\\')
-    string = string.replace('"', '\\"')
-    string = string.replace('\t', '\\t')
-    string = string.replace('\n', '\\n')
-
-    return string
-
   @staticmethod
   def _get_regex_for_type(type_):
     matches = filter(lambda field_type: field_type.name == type_, FIELD_TYPES)
@@ -179,10 +170,10 @@ class Indexer(object):
       "collection_name":collection_name,
       "fields":self.get_field_list(data['columns']),
       "num_base_fields": len(data['columns']),
-      "format_character":Indexer._format_character,
       "uuid_name" : uuid_name,
       "get_regex":Indexer._get_regex_for_type,
-      "format":data['format'],
+      "format_settings": data['format'],
+      "format_class": get_file_format_class(data['format']['type']),
       "get_kept_args": get_checked_args,
       "grok_dictionaries_location" : grok_dicts_loc if self.fs.exists(grok_dicts_loc) else None,
       "geolite_db_location" : geolite_loc if self.fs.exists(geolite_loc) else None,