Adds configurable JSON log parsing(zalando-incubator#49)

With the new configuration parameter `WATCHER_SCALYR_PARSE_LINES_JSON` the Scalyr agent can be configured to parse log lines as JSON entries which is useful to process raw Docker logs. This behaviour is disabled by default.

Author: eicnix

README.rst

@@ -291,6 +291,9 @@ WATCHER_SCALYR_DEST_PATH
 WATCHER_SCALYR_CONFIG_PATH
   Scalyr configuration file path. (Default: ``/etc/scalyr-agent-2/agent.json``)
 
+WATCHER_SCALYR_PARSE_LINES_JSON
+  Parse lines lines on the client as JSON. Useful for raw docker logs. (Default: ``False``)
+
 WATCHER_SCALYR_JOURNALD
   Scalyr should follow Journald logs. This is for node system processes log shipping (e.g. docker, kube) (Default: ``False``)
 

kube_log_watcher/agents/scalyr.py

@@ -33,6 +33,7 @@ def __init__(self, cluster_id: str, load_template):
         self.api_key = os.environ.get('WATCHER_SCALYR_API_KEY')
         self.dest_path = os.environ.get('WATCHER_SCALYR_DEST_PATH')
         self.scalyr_server = os.environ.get('WATCHER_SCALYR_SERVER')
+        self.parse_lines_json = os.environ.get('WATCHER_SCALYR_PARSE_LINES_JSON', False)
 
         if not all([self.api_key, self.dest_path]):
             raise RuntimeError('Scalyr watcher agent initialization failed. Env variables WATCHER_SCALYR_API_KEY and '
@@ -205,6 +206,7 @@ def flush(self):
             'logs': self.logs,
             'monitor_journald': self.journald,
             'scalyr_server': self.scalyr_server,
+            'parse_lines_json': self.parse_lines_json,
         }
 
         current_paths = self._get_current_log_paths()

kube_log_watcher/templates/scalyr.json.jinja2

@@ -25,6 +25,10 @@
                 "redaction_rules": {{ log.redaction_rules | tojson }},
                 {% endif %}
 
+                {% if parse_lines_json %}
+                "parse_lines_as_json": true,
+                {% endif %}
+
                 "copy_from_start": true,
 
                 "attributes": {{ log.attributes | tojson }}

tests/conftest.py

@@ -202,6 +202,7 @@ def fx_containers_sync(request):
     'scalyr_key': SCALYR_KEY,
     'cluster_id': CLUSTER_ID,
     'monitor_journald': None,
+    'parse_lines_json': False,
     'logs': [
         {
             'path': os.path.join(SCALYR_DEST_PATH, 'container-1', 'app-1-v1.log'),

tests/test_scalyr.py

@@ -31,7 +31,7 @@
     },
 )
 
-KWARGS_KEYS = ('scalyr_key', 'cluster_id', 'logs', 'monitor_journald')
+KWARGS_KEYS = ('scalyr_key', 'parse_lines_json', 'cluster_id', 'logs', 'monitor_journald')
 
 
 SCALYR_MONITOR_JOURNALD = copy.deepcopy(SCALYR_JOURNALD_DEFAULTS)
@@ -489,6 +489,38 @@ def test_remove_log_target(monkeypatch, env, exc):
                 ],
             },
         ),
+        (
+                {
+                    'scalyr_key': SCALYR_KEY,
+                    'cluster_id': CLUSTER_ID,
+                    'parse_lines_json': True,
+                    'monitor_journald': None,
+                    'logs': [
+                        {
+                            'path': '/p1',
+                            'attributes': {'a1': 'v1', 'parser': 'c-parser'},
+                            'copy_from_start': True,
+                            'redaction_rules': {'match_expression': 'match-expression'}
+                        }
+                    ]
+                },
+                {
+                    'api_key': 'scalyr-key-123',
+                    'implicit_metric_monitor': False,
+                    'implicit_agent_process_metrics_monitor': False,
+                    'server_attributes': {'serverHost': 'kube-cluster'},
+                    'monitors': [],
+                    'logs': [
+                        {
+                            'attributes': {'a1': 'v1', 'parser': 'c-parser'},
+                            'path': '/p1',
+                            'parse_lines_as_json': True,
+                            'copy_from_start': True,
+                            'redaction_rules': {'match_expression': 'match-expression'}
+                        }
+                    ],
+                },
+        ),
     )
 )
 def test_tpl_render(monkeypatch, kwargs, expected):