Repository: zalando-incubator/kubernetes-log-watcher

Issue:
Persist checkpoint file
Scalyr agent uses a checkpoint file to avoid log duplication. This should be highlighted/used in a Kubernetes cluster since the file should survive agent container restarts.

Discussion:
codecov-io: # [Codecov](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=h1) Report
> Merging [#32](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=desc) into [master](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/commit/7b9c8c63c94785b04f9e4c6509d6cfeb6876763c?src=pr&el=desc) will **decrease** coverage by `0.06%`.
> The diff coverage is `75%`.

[![Impacted file tree graph](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32/graphs/tree.svg?src=pr&height=150&token=2NLvIPFi70&width=650)](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##           master     #32      +/-   ##
=========================================
- Coverage   78.97%   78.9%   -0.07%     
=========================================
  Files           9       9              
  Lines         371     384      +13     
=========================================
+ Hits          293     303      +10     
- Misses         78      81       +3
```


| [Impacted Files](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [kube\_log\_watcher/kube.py](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=tree#diff-a3ViZV9sb2dfd2F0Y2hlci9rdWJlLnB5) | `69.23% <0%> (-3.75%)` | :arrow_down: |
| [kube\_log\_watcher/agents/scalyr.py](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=tree#diff-a3ViZV9sb2dfd2F0Y2hlci9hZ2VudHMvc2NhbHlyLnB5) | `92.3% <88.23%> (-0.17%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=footer). Last update [7b9c8c6...0b31a74](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/32?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
lmineiro: Although I can't, in all honesty, grasp parts of this, it LGTM
lmineiro: 👍 
mohabusama: 👍 

Patch:
diff --git a/README.rst b/README.rst
index 64a7b52..2a6ae78 100644
--- a/README.rst
+++ b/README.rst
@@ -89,34 +89,16 @@ Operation
 Example manifest
 ----------------
 
-This is an example manifest for shipping logs to Scalyr, with additional Journald monitoring for master processes running on the node:
+This is an example manifest for shipping logs to Scalyr, with additional Journald monitoring for master processes running on the node.
 
-.. code-block:: yaml
-
-    apiVersion: v1
-    kind: ConfigMap
-    metadata:
-        name: scalyr-config-base
-        namespace: kube-system
-    data:
-        scalyr.config: |
-          {
-            "import_vars": ["WATCHER_SCALYR_API_KEY", "WATCHER_CLUSTER_ID"],
+.. note::
 
-            "api_key": "$WATCHER_SCALYR_API_KEY",
+    - This manifest assumes running a Kubernetes cluster version > 1.5 (as it depends on `initContainer <https://kubernetes.io/docs/concepts/workloads/pods/init-containers/>`_ for initial Scalyr configuration)
+    - All shared volumes are of type ``hostPath`` in order to survive pod restarts.
+    - Initial Scalyr configuration using ``configMap`` is no longer used as it appears to be reseted to initial values by Kubernetes.
 
-            "server_attributes": {
-                "serverHost": "$WATCHER_CLUSTER_ID"
-            },
-
-            "implicit_metric_monitor": false,
-            "implicit_agent_process_metrics_monitor": false,
-
-            "logs": [],
-            "monitors": []
-          }
+.. code-block:: yaml
 
-    ---
     apiVersion: extensions/v1beta1
     kind: DaemonSet
     metadata:
@@ -140,18 +122,50 @@ This is an example manifest for shipping logs to Scalyr, with additional Journal
             annotations:
               scheduler.alpha.kubernetes.io/critical-pod: ''
               scheduler.alpha.kubernetes.io/tolerations: '[{"key":"CriticalAddonsOnly", "operator":"Exists"}]'
+              pod.beta.kubernetes.io/init-containers: '[
+                {
+                  "name": "init-scalyr-config",
+                  "image": "busybox",
+                  "imagePullPolicy": "IfNotPresent",
+                  "command": ["sh", "-c"],
+                  "args": [
+                    "if [ ! -f /mnt/scalyr/agent.json ]; then
+                      echo {
+                        \\\"import_vars\\\": [\\\"WATCHER_SCALYR_API_KEY\\\", \\\"WATCHER_CLUSTER_ID\\\"],
+                        \\\"server_attributes\\\": {\\\"serverHost\\\": \\\"\\$WATCHER_CLUSTER_ID\\\"},
+                        \\\"implicit_agent_process_metrics_monitor\\\": false,
+                        \\\"implicit_metric_monitor\\\": false,
+                        \\\"api_key\\\": \\\"\\$WATCHER_SCALYR_API_KEY\\\",
+                        \\\"monitors\\\": [],
+                        \\\"logs\\\": []
+                        } > /mnt/scalyr/agent.json;
+                        echo Updated agent.json to inital configuration;
+                    fi
+                    && cat /mnt/scalyr/agent.json;
+                    test -f /mnt/scalyr-checkpoint/checkpoints.json && ls -lah /mnt/scalyr-checkpoint/checkpoints.json && cat /mnt/scalyr-checkpoint/checkpoints.json"
+                  ],
+                  "volumeMounts": [
+                    {
+                      "name": "scalyr-config",
+                      "mountPath": "/mnt/scalyr"
+                    },
+                    {
+                      "name": "scalyr-checkpoint",
+                      "mountPath": "/mnt/scalyr-checkpoint"
+                    }
+                  ]
+                }
+              ]'
           spec:
             containers:
             - name: log-watcher
-              image: registry.opensource.zalan.do/eagleeye/kubernetes-log-watcher:0.11
+              image: registry.opensource.zalan.do/eagleeye/kubernetes-log-watcher:0.12
               env:
               - name: CLUSTER_NODE_NAME
                 valueFrom:
                   fieldRef:
                     fieldPath: spec.nodeName
 
-              - name: WATCHER_KUBERNETES_UPDATE_CERTIFICATES
-                value: "true"
               - name: WATCHER_DEBUG
                 value: "true"
               - name: WATCHER_CLUSTER_ID
@@ -177,13 +191,13 @@ This is an example manifest for shipping logs to Scalyr, with additional Journal
                 readOnly: false
               - name: scalyr-config
                 mountPath: /mnt/scalyr-config
-                readOnly: false
 
             - name: scalyr-agent
-              image: registry.opensource.zalan.do/eagleeye/scalyr-agent:0.1
+
+              image: registry.opensource.zalan.do/eagleeye/scalyr-agent:0.2
 
               env:
-              # Note: added for scalyr-config-base, but not needed by the scalyr-agent itself.
+              # Note: added for scalyr-config, but not needed by the scalyr-agent itself.
               - name: WATCHER_SCALYR_API_KEY
                 value: "<SCALYR-KEY-HERE>"
               - name: WATCHER_CLUSTER_ID
@@ -196,8 +210,10 @@ This is an example manifest for shipping logs to Scalyr, with additional Journal
               - name: scalyr-logs
                 mountPath: /mnt/scalyr-logs
                 readOnly: true
+              - name: scalyr-checkpoint
+                mountPath: /var/lib/scalyr-agent-2
               - name: scalyr-config
-                mountPath: /etc/scalyr-agent-2/
+                mountPath: /etc/scalyr-agent-2
                 readOnly: true
               - name: journal
                 mountPath: /var/log/journal
@@ -212,15 +228,17 @@ This is an example manifest for shipping logs to Scalyr, with additional Journal
               hostPath:
                 path: /var/log/journal
 
-            - name: scalyr-logs
-              emptyDir: {}
+            - name: scalyr-checkpoint
+              hostPath:
+                path: /var/lib/scalyr-agent
 
             - name: scalyr-config
-              configMap:
-                name: scalyr-config-base
-                items:
-                  - key: scalyr.config
-                    path: agent.json
+              hostPath:
+                path: /etc/scalyr-agent
+
+            - name: scalyr-logs
+              hostPath:
+                path: /var/log/scalyr-agent
 
 
 Configuration
diff --git a/kube_log_watcher/agents/scalyr.py b/kube_log_watcher/agents/scalyr.py
index ad831fe..798d229 100644
--- a/kube_log_watcher/agents/scalyr.py
+++ b/kube_log_watcher/agents/scalyr.py
@@ -31,6 +31,15 @@ class ScalyrAgent(BaseWatcher):
                 'Scalyr watcher agent initialization failed. {} config path does not exist.'.format(
                     self.config_path))
 
+        if not os.path.exists(self.dest_path):
+            raise RuntimeError(
+                'Scalyr watcher agent initialization failed. {} destination path does not exist.'.format(
+                    self.dest_path))
+        else:
+            watched_containers = os.listdir(self.dest_path)
+            logger.info('Scalyr watcher agent found {} watched containers.'.format(len(watched_containers)))
+            logger.debug('Scalyr watcher agent found the following watched containers: {}'.format(watched_containers))
+
         self.journald = None
         journald_monitor = os.environ.get('WATCHER_SCALYR_JOURNALD', False)
 
@@ -109,7 +118,11 @@ class ScalyrAgent(BaseWatcher):
         current_paths = self._get_current_log_paths()
         new_paths = {log['path'] for log in self.logs}
 
-        if self._first_run or new_paths.symmetric_difference(current_paths):
+        diff_paths = new_paths.symmetric_difference(current_paths)
+
+        if self._first_run or diff_paths:
+            logger.debug('Scalyr watcher agent new paths: {}'.format(diff_paths))
+            logger.debug('Scalyr watcher agent current paths: {}'.format(current_paths))
             try:
                 config = self.tpl.render(**kwargs)
 
@@ -119,7 +132,8 @@ class ScalyrAgent(BaseWatcher):
                 logger.exception('Scalyr watcher agent failed to write config file.')
             else:
                 self._first_run = False
-                logger.info('Scalyr watcher agent updated config file {}'.format(self.config_path))
+                logger.info('Scalyr watcher agent updated config file {} with {} log targets.'.format(
+                    self.config_path, len(diff_paths)))
 
     def reset(self):
         self.logs = []
@@ -155,10 +169,15 @@ class ScalyrAgent(BaseWatcher):
         targets = set()
 
         try:
-            with open(self.config_path) as fp:
-                config = json.load(fp)
-                targets = {log.get('path') for log in config.get('logs', [])}
+            if os.path.exists(self.config_path):
+                with open(self.config_path) as fp:
+                    config = json.load(fp)
+                    targets = {log.get('path') for log in config.get('logs', [])}
+                    logger.debug('Scalyr watcher agent loaded existing config {}: {} log targets exist!'.format(
+                        self.config_path, len(config.get('logs', []))))
+            else:
+                logger.warning('Scalyr watcher agent cannot find config file!')
         except:
-            pass
+            logger.exception('Scalyr watcher agent failed to read config!')
 
         return targets
diff --git a/kube_log_watcher/kube.py b/kube_log_watcher/kube.py
index ac312ab..6db5d05 100644
--- a/kube_log_watcher/kube.py
+++ b/kube_log_watcher/kube.py
@@ -30,7 +30,10 @@ def update_ca_certificate():
 
 def get_client():
     config = pykube.KubeConfig.from_service_account(DEFAULT_SERVICE_ACC)
-    return pykube.HTTPClient(config)
+    client = pykube.HTTPClient(config)
+    client.session.trust_env = False
+
+    return client
 
 
 def get_pods(kube_url=None, namespace=DEFAULT_NAMESPACE) -> list:
diff --git a/kube_log_watcher/templates/scalyr.json.jinja2 b/kube_log_watcher/templates/scalyr.json.jinja2
index b43c864..58abce4 100644
--- a/kube_log_watcher/templates/scalyr.json.jinja2
+++ b/kube_log_watcher/templates/scalyr.json.jinja2
@@ -13,6 +13,8 @@
             {
                 "path": "{{ log.path }}",
 
+                "copy_from_start": true,
+
                 "attributes": {
                     "parser": "json"{% if log.attributes %},{% endif %}
                     {% for k, v in log.attributes.items() %}
@@ -45,7 +47,9 @@
             },
             {% endif %}
 
-            "module": "scalyr_agent.builtin_monitors.journald_monitor"
+            "module": "scalyr_agent.builtin_monitors.journald_monitor",
+            "monitor_log_write_rate": 10000,
+            "monitor_log_max_write_burst": 200000
         }
         {% endif %}
     ]
diff --git a/requirements.txt b/requirements.txt
index 9b2b676..23b4c8a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,2 +1,2 @@
 Jinja2==2.8
-pykube==0.13.0
+pykube>=0.15.0


Test patch:
diff --git a/tests/test_scalyr.py b/tests/test_scalyr.py
index 1af390c..ad96269 100644
--- a/tests/test_scalyr.py
+++ b/tests/test_scalyr.py
@@ -61,11 +61,14 @@ def patch_env(monkeypatch, env):
 def patch_os(monkeypatch):
     makedirs = MagicMock()
     symlink = MagicMock()
+    listdir = MagicMock()
+    listdir.return_value = []
 
     monkeypatch.setattr('os.makedirs', makedirs)
     monkeypatch.setattr('os.symlink', symlink)
+    monkeypatch.setattr('os.listdir', listdir)
 
-    return makedirs, symlink
+    return makedirs, symlink, listdir
 
 
 def patch_open(monkeypatch, exc=None):
@@ -81,6 +84,54 @@ def patch_open(monkeypatch, exc=None):
     return mock_open, mock_fp
 
 
+@pytest.mark.parametrize(
+    'env,exists',
+    (
+        (
+            {
+                # No API KEY
+                'WATCHER_SCALYR_DEST_PATH': SCALYR_DEST_PATH,
+                'WATCHER_SCALYR_CONFIG_PATH': '/etc/config'
+            },
+            (True, True)
+        ),
+        (
+            {
+                # No Dest path
+                'WATCHER_SCALYR_API_KEY': SCALYR_KEY,
+            },
+            (True, True)
+        ),
+        (
+            {
+                'WATCHER_SCALYR_API_KEY': SCALYR_KEY, 'WATCHER_SCALYR_DEST_PATH': SCALYR_DEST_PATH,
+                'WATCHER_SCALYR_CONFIG_PATH': '/etc/config'
+            },
+            # Config path does not exist
+            (False, True)
+        ),
+        (
+            {
+                'WATCHER_SCALYR_API_KEY': SCALYR_KEY, 'WATCHER_SCALYR_DEST_PATH': SCALYR_DEST_PATH,
+                'WATCHER_SCALYR_CONFIG_PATH': '/etc/config'
+            },
+            # Dest path does not exist
+            (True, False)
+        ),
+    )
+)
+def test_initialization_failure(monkeypatch, env, exists):
+    patch_env(monkeypatch, env)
+    patch_os(monkeypatch)
+
+    exists = MagicMock()
+    exists.side_effect = exists
+    monkeypatch.setattr('os.path.exists', exists)
+
+    with pytest.raises(RuntimeError):
+        ScalyrAgent(CLUSTER_ID, load_template)
+
+
 @pytest.mark.parametrize('env', ENVS)
 def test_add_log_target(monkeypatch, env, fx_scalyr):
     patch_env(monkeypatch, env)
@@ -94,10 +145,10 @@ def test_add_log_target(monkeypatch, env, fx_scalyr):
     kwargs['monitor_journald'] = {} if not env.get('WATCHER_SCALYR_JOURNALD') else SCALYR_MONITOR_JOURNALD
 
     exists = MagicMock()
-    exists.side_effect = (True, True, False, False)
+    exists.side_effect = (True, True, True, False, False, True)
     monkeypatch.setattr('os.path.exists', exists)
 
-    makedirs, symlink = patch_os(monkeypatch)
+    makedirs, symlink, listdir = patch_os(monkeypatch)
 
     current_targets = MagicMock()
     current_targets.return_value = []
@@ -125,12 +176,13 @@ def test_add_log_target(monkeypatch, env, fx_scalyr):
 
 @pytest.mark.parametrize('env', ENVS)
 def test_add_log_target_no_src(monkeypatch, env, fx_scalyr):
+    patch_os(monkeypatch)
     patch_env(monkeypatch, env)
 
     target = fx_scalyr['target']
 
     exists = MagicMock()
-    exists.side_effect = (True, False)
+    exists.side_effect = (True, True, False)
     monkeypatch.setattr('os.path.exists', exists)
 
     agent = ScalyrAgent(CLUSTER_ID, load_template)
@@ -155,10 +207,10 @@ def test_add_log_target_no_change(monkeypatch, env, fx_scalyr):
     kwargs['monitor_journald'] = {} if not env.get('WATCHER_SCALYR_JOURNALD') else SCALYR_MONITOR_JOURNALD
 
     exists = MagicMock()
-    exists.side_effect = (True, True, False, False)
+    exists.side_effect = (True, True, True, False, False, True)
     monkeypatch.setattr('os.path.exists', exists)
 
-    makedirs, symlink = patch_os(monkeypatch)
+    makedirs, symlink, listdir = patch_os(monkeypatch)
 
     log_path = kwargs['logs'][0]['path']
 
@@ -200,10 +252,10 @@ def test_flush_failure(monkeypatch, env, fx_scalyr):
     kwargs['monitor_journald'] = {} if not env.get('WATCHER_SCALYR_JOURNALD') else SCALYR_MONITOR_JOURNALD
 
     exists = MagicMock()
-    exists.side_effect = (True, True, False, False)
+    exists.side_effect = (True, True, True, False, False, True)
     monkeypatch.setattr('os.path.exists', exists)
 
-    makedirs, symlink = patch_os(monkeypatch)
+    makedirs, symlink, listdir = patch_os(monkeypatch)
 
     log_path = kwargs['logs'][0]['path']
 
@@ -250,10 +302,10 @@ def test_get_current_log_paths(monkeypatch, env, config, result):
     monkeypatch.setattr('json.load', load)
 
     exists = MagicMock()
-    exists.side_effect = (True, True, False, False)
+    exists.side_effect = (True, True, True, False, False, True)
     monkeypatch.setattr('os.path.exists', exists)
 
-    makedirs, symlink = patch_os(monkeypatch)
+    makedirs, symlink, listdir = patch_os(monkeypatch)
 
     agent = ScalyrAgent(CLUSTER_ID, load_template)
 
@@ -280,10 +332,11 @@ def test_get_current_log_paths(monkeypatch, env, config, result):
     )
 )
 def test_remove_log_target(monkeypatch, env, exc):
+    patch_os(monkeypatch)
     patch_env(monkeypatch, env)
 
     exists = MagicMock()
-    exists.side_effect = (True, True, False, False)
+    exists.side_effect = (True, True, True, False, False, True)
     monkeypatch.setattr('os.path.exists', exists)
 
     rmtree = MagicMock()
@@ -328,6 +381,8 @@ def test_remove_log_target(monkeypatch, env, exc):
                 'monitors': [
                     {
                         'module': 'scalyr_agent.builtin_monitors.journald_monitor',
+                        'monitor_log_write_rate': 10000,
+                        'monitor_log_max_write_burst': 200000,
                     }
                 ]
             },
@@ -335,7 +390,7 @@ def test_remove_log_target(monkeypatch, env, exc):
         (
             {
                 'scalyr_key': SCALYR_KEY, 'cluster_id': CLUSTER_ID,
-                'logs': [{'path': '/p1', 'attributes': {'a1': 'v1'}}],
+                'logs': [{'path': '/p1', 'attributes': {'a1': 'v1'}, 'copy_from_start': True}],
                 'monitor_journald': {
                     'journal_path': '/var/log/journal',
                     'attributes': {'cluster': CLUSTER_ID, 'node': NODE},
@@ -347,10 +402,12 @@ def test_remove_log_target(monkeypatch, env, exc):
                 'implicit_metric_monitor': False,
                 'implicit_agent_process_metrics_monitor': False,
                 'server_attributes': {'serverHost': 'kube-cluster'},
-                'logs': [{'attributes': {'a1': 'v1', 'parser': 'json'}, 'path': '/p1'}],
+                'logs': [{'attributes': {'a1': 'v1', 'parser': 'json'}, 'path': '/p1', 'copy_from_start': True}],
                 'monitors': [
                     {
                         'module': 'scalyr_agent.builtin_monitors.journald_monitor',
+                        'monitor_log_write_rate': 10000,
+                        'monitor_log_max_write_burst': 200000,
                         'journal_path': '/var/log/journal',
                         'attributes': {'cluster': CLUSTER_ID, 'node': NODE},
                         'extra_fields': {'_COMM': 'command'}