Allow sampling and redaction rules for Scalyr (zalando-incubator#62)

Felix Mueller · mohabusama · commit ceb6139ce93b · 2018-06-12T10:55:27.000+02:00
* Allow sampling and redaction rules for Scalyr This feature enables using sampling and redaction rules for Scalyr. More details about sampling_rules and redaction_rules can be found here: https://eu.scalyr.com/help/scalyr-agent * Updating requirements.txt * fixing flake8 errors * Updating readme * Updated README
diff --git a/README.rst b/README.rst
@@ -321,6 +321,29 @@ The default parser for container logs is ``json`` parser. In some cases however
 
 The value of ``kubernetes-log-watcher/scalyr-parser`` annotation should be a json serialized list. If ``container`` value did not match, then default parser is used (i.e. ``json``).
 
+Scalyr sampling rules
+....................
+
+Sampling rules enable to only ship a certain pattern that matches a regular expression and specified amount of log percentage to Scalyr. The example shows an expression that matches ``app-1`` and a match expression ``my-expression``. If it's met, only 10% of it will be shipped to Scalyr using a ``sampling_rate`` of ``0.1``.
+
+.. code-block:: yaml
+
+  annotations:
+    kubernetes-log-watcher/scalyr-sampling-rules: '[{"container": "app-1", "sampling-rules":[{ "match_expression": "my-expression", "sampling_rate": "0.1" }]}]'
+
+
+Scalyr log redaction
+....................
+
+Redaction rules enable to avoid shipping sensitive data that shouldn't get transferred to Scalyr either getting fully removed from log files or to replace them with specific strings. The first example below shows how matches will be fully removed and the second shows how matches will be replaced with a different string.
+
+.. code-block:: yaml
+
+  annotations:
+    kubernetes-log-watcher/scalyr-redaction-rules: '[{"container": "app-1", "redaction-rules":[{ "match_expression": "my-expression" }]}]'
+    kubernetes-log-watcher/scalyr-redaction-rules: '[{"container": "app-1", "redaction-rules":[{ "match_expression": "my-expression", "replacement": "replacement-expression" }]}]'
+
+
 AppDynamics configuration agent
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
diff --git a/kube_log_watcher/agents/scalyr.py b/kube_log_watcher/agents/scalyr.py
@@ -15,6 +15,12 @@
 
 # If exists! we expect serialized json str: '[{"container": "my-container", "parser": "my-custom-parser"}]'
 SCALYR_ANNOTATION_PARSER = 'kubernetes-log-watcher/scalyr-parser'
+# If exists! we expect serialized json str:
+# '[{"container": "my-container", "sampling-rules":[{ "match_expression": "<expression here>",
+#  "sampling_rate": "0" }]}]'
+SCALYR_ANNOTATION_SAMPLING_RULES = 'kubernetes-log-watcher/scalyr-sampling-rules'
+# '[{"container": "my-container", "redaction-rules":[{ "match_expression": "<expression here>" }]}]'
+SCALYR_ANNOTATION_REDACTION_RULES = 'kubernetes-log-watcher/scalyr-redaction-rules'
 SCALYR_DEFAULT_PARSER = 'json'
 SCALYR_DEFAULT_WRITE_RATE = 10000
 SCALYR_DEFAULT_WRITE_BURST = 200000
@@ -89,6 +95,8 @@ def add_log_target(self, target: dict):
             return
 
         parser = SCALYR_DEFAULT_PARSER
+        sampling_rules = None
+        redaction_rules = None
 
         annotations = target['kwargs'].get('pod_annotations', {})
         if annotations and SCALYR_ANNOTATION_PARSER in annotations:
@@ -105,11 +113,64 @@ def add_log_target(self, target: dict):
                             parser = p.get('parser', SCALYR_DEFAULT_PARSER)
                             logger.debug('Scalyr watcher agent loaded parser: {} for container: {}'.format(
                                 parser, target['kwargs']['container_name']))
+                            break
             except Exception:
                 logger.error('Scalyr watcher agent failed to load annotation {}'.format(SCALYR_ANNOTATION_PARSER))
 
+        if annotations and SCALYR_ANNOTATION_SAMPLING_RULES in annotations:
+            try:
+                containers_sampling_rules = json.loads(annotations[SCALYR_ANNOTATION_SAMPLING_RULES])
+                if type(containers_sampling_rules) is not list:
+                    logger.warning(
+                        ('Scalyr watcher agent found invalid {} annotation in pod: {}. '
+                         'Expected `list` found: `{}`').format(
+                            SCALYR_ANNOTATION_SAMPLING_RULES,
+                            target['kwargs']['pod_name'], type(containers_sampling_rules)))
+                else:
+                    for p in containers_sampling_rules:
+                        if p.get('container') == target['kwargs']['container_name']:
+                            sampling_rules = p.get('sampling-rules')
+                            if not sampling_rules:
+                                logger.warning(
+                                    ('Scalyr watcher agent did not find sampling rules for {}').format(
+                                     target['kwargs']['container_name']))
+                            else:
+                                logger.debug('Scalyr watcher agent loaded sampling-rule: {} for container: {}'.format(
+                                    sampling_rules, target['kwargs']['container_name']))
+                                break
+            except Exception:
+                logger.error('Scalyr watcher agent failed to load annotation {}'.format
+                             (SCALYR_ANNOTATION_SAMPLING_RULES))
+
+        if annotations and SCALYR_ANNOTATION_REDACTION_RULES in annotations:
+            try:
+                containers_redaction_rules = json.loads(annotations[SCALYR_ANNOTATION_REDACTION_RULES])
+                if type(containers_redaction_rules) is not list:
+                    logger.warning(
+                        ('Scalyr watcher agent found invalid {} annotation in pod: {}. '
+                         'Expected `list` found: `{}`').format(
+                            SCALYR_ANNOTATION_REDACTION_RULES,
+                            target['kwargs']['pod_name'], type(containers_redaction_rules)))
+                else:
+                    for p in containers_redaction_rules:
+                        if p.get('container') == target['kwargs']['container_name']:
+                            redaction_rules = p.get('redaction-rules')
+                            if not redaction_rules:
+                                logger.warning(
+                                    ('Scalyr watcher agent did not find redaction rules for {}').format(
+                                     target['kwargs']['container_name']))
+                            else:
+                                logger.debug('Scalyr watcher agent loaded redaction-rule: {} for container: {}'.format(
+                                    redaction_rules, target['kwargs']['container_name']))
+                                break
+            except Exception:
+                logger.error('Scalyr watcher agent failed to load annotation {}'.format
+                             (SCALYR_ANNOTATION_REDACTION_RULES))
+
         log = {
             'path': log_path,
+            'sampling_rules': sampling_rules,
+            'redaction_rules': redaction_rules,
             'attributes': {
                 'application': target['kwargs']['application_id'],
                 'version': target['kwargs']['application_version'],
diff --git a/kube_log_watcher/templates/scalyr.json.jinja2 b/kube_log_watcher/templates/scalyr.json.jinja2
@@ -17,13 +17,17 @@
             {
                 "path": "{{ log.path }}",
 
+                {% if log.sampling_rules %}
+                "sampling_rules": {{ log.sampling_rules | tojson }},
+                {% endif %}
+
+                {% if log.redaction_rules %}
+                "redaction_rules": {{ log.redaction_rules | tojson }},
+                {% endif %}
+
                 "copy_from_start": true,
 
-                "attributes": {
-                    {% for k, v in log.attributes.items() %}
-                    "{{ k }}": "{{ v }}"{% if not loop.last %},{% endif %}
-                    {% endfor %}
-                }
+                "attributes": {{ log.attributes | tojson }}
             }{% if not loop.last %},{% endif %}
         {% endfor %}
     ],
diff --git a/requirements.txt b/requirements.txt
@@ -1,2 +1,2 @@
-Jinja2==2.8
+Jinja2
 pykube>=0.15.0
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -15,6 +15,8 @@
     'write_burst': 200000
 }
 SCALYR_ANNOTATION_PARSER = 'kubernetes-log-watcher/scalyr-parser'
+SCALYR_ANNOTATION_SAMPLING_RULES = 'kubernetes-log-watcher/scalyr-sampling-rules'
+SCALYR_ANNOTATION_REDACTION_RULES = 'kubernetes-log-watcher/scalyr-redaction-rules'
 
 TARGET = {
     'id': 'container-1',
@@ -31,7 +33,14 @@
         'container_id': 'container-1',
         'container_path': '/mnt/containers/container-1',
         'log_file_name': 'container-1-json.log',
-        'pod_annotations': {SCALYR_ANNOTATION_PARSER: '[{"container": "app-1-container-1", "parser": "custom-parser"}]'}
+        'pod_annotations': {
+            SCALYR_ANNOTATION_PARSER: '[{"container": "app-1-container-1", "parser": "custom-parser"}]',
+            SCALYR_ANNOTATION_SAMPLING_RULES:
+                '[{"container": "app-1-container-1", "sampling-rules":[{ "match_expression": "<expression here>", '
+                '"sampling_rate": "0" }]}]',
+            SCALYR_ANNOTATION_REDACTION_RULES:
+                '[{"container": "app-1-container-1", "redaction-rules":[{ "match_expression": "<expression here>" }]}]'
+        }
 
     },
     'pod_labels': {}
diff --git a/tests/test_scalyr.py b/tests/test_scalyr.py
@@ -429,6 +429,66 @@ def test_remove_log_target(monkeypatch, env, exc):
                 ]
             },
         ),
+        (
+            {
+                'scalyr_key': SCALYR_KEY,
+                'cluster_id': CLUSTER_ID,
+                'monitor_journald': None,
+                'logs': [
+                    {
+                        'path': '/p1',
+                        'attributes': {'a1': 'v1', 'parser': 'c-parser'},
+                        'copy_from_start': True,
+                        'sampling_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': [
+                    {
+                        'path': '/p1',
+                        'attributes': {'a1': 'v1', 'parser': 'c-parser'},
+                        'copy_from_start': True,
+                        'sampling_rules': {'match_expression': 'match-expression'}
+                    }
+                ],
+            },
+        ),
+        (
+            {
+                'scalyr_key': SCALYR_KEY,
+                'cluster_id': CLUSTER_ID,
+                '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',
+                        'copy_from_start': True,
+                        'redaction_rules': {'match_expression': 'match-expression'}
+                    }
+                ],
+            },
+        ),
     )
 )
 def test_tpl_render(monkeypatch, kwargs, expected):

Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
`1`		`-Jinja2==2.8`
	`1`	`+Jinja2`
`2`	`2`	`pykube>=0.15.0`