Repository: zalando-incubator/kubernetes-log-watcher

Issue:
Scalyr: pod labels and annotations extraction fail
The issue is inconsistent, but in some pods created by cronjobs, log-watcher fails to get the ``labels`` and ``annotations`` which lead to breaking custom parsing and log search based on ``application`` or ``version`` fields.

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

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

```diff
@@            Coverage Diff             @@
##           master      #61      +/-   ##
==========================================
+ Coverage   84.68%   85.04%   +0.35%     
==========================================
  Files           9        9              
  Lines         418      428      +10     
==========================================
+ Hits          354      364      +10     
  Misses         64       64
```


| [Impacted Files](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/61?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [kube\_log\_watcher/kube.py](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/61/diff?src=pr&el=tree#diff-a3ViZV9sb2dfd2F0Y2hlci9rdWJlLnB5) | `83.72% <100%> (+0.38%)` | :arrow_up: |
| [kube\_log\_watcher/main.py](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/61/diff?src=pr&el=tree#diff-a3ViZV9sb2dfd2F0Y2hlci9tYWluLnB5) | `77.52% <100%> (+1.19%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/61?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/61?src=pr&el=footer). Last update [c13fd09...19b7fe6](https://codecov.io/gh/zalando-incubator/kubernetes-log-watcher/pull/61?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

vetinari: :+1:
vetinari: :+1:
mohabusama: 👍 

Patch:
diff --git a/delivery.yaml b/delivery.yaml
index 26e68f7..bb7cad7 100644
--- a/delivery.yaml
+++ b/delivery.yaml
@@ -1,6 +1,7 @@
 build_steps:
 - desc: "Install dependencies"
   cmd: |
+    apt-get update
     apt-get install -q -y --no-install-recommends \
       git \
       python3.5 \
@@ -28,7 +29,7 @@ build_steps:
       RELEASE_VERSION=$(git describe --tags --always --dirty)
       AGENT_IMAGE=registry-write.opensource.zalan.do/eagleeye/kubernetes-log-watcher:${RELEASE_VERSION}
     else
-      AGENT_IMAGE=registry-write.opensource.zalan.do/eagleeye/kubernetes-log-watcher:${CDP_BUILD_VERSION}
+      AGENT_IMAGE=registry-write.opensource.zalan.do/eagleeye/kubernetes-log-watcher-unstable:${CDP_BUILD_VERSION}
     fi
     docker build --tag "$AGENT_IMAGE" .
     docker push "$AGENT_IMAGE"
diff --git a/kube_log_watcher/kube.py b/kube_log_watcher/kube.py
index 73c1ffc..fbfc095 100644
--- a/kube_log_watcher/kube.py
+++ b/kube_log_watcher/kube.py
@@ -6,8 +6,6 @@ import warnings
 
 from urllib.parse import urljoin
 
-from typing import Tuple
-
 import pykube
 import requests
 
@@ -15,13 +13,17 @@ import requests
 DEFAULT_SERVICE_ACC = '/var/run/secrets/kubernetes.io/serviceaccount'
 DEFAULT_NAMESPACE = 'default'
 
-PODS_URL = 'api/v1/namespaces/{}/pods'
+PODS_URL = 'api/v1/namespaces/{}/pods/{}'
 
 PAUSE_CONTAINER_PREFIX = 'gcr.io/google_containers/pause-'
 
 logger = logging.getLogger('kube_log_watcher')
 
 
+class PodNotFound(Exception):
+    pass
+
+
 def update_ca_certificate():
     warnings.warn('update_ca_certificate is deprecated.')
     try:
@@ -40,39 +42,37 @@ def get_client():
     return client
 
 
-def get_pods(kube_url=None, namespace=DEFAULT_NAMESPACE) -> list:
+def get_pod(name, namespace=DEFAULT_NAMESPACE, kube_url=None) -> pykube.Pod:
     """
-    Return list of pods in cluster.
+    Return Pod with name.
     If ``kube_url`` is not ``None`` then kubernetes service account config won't be used.
 
+    :param name: Pod name to use in filtering.
+    :type name: str
+
+    :param namespace: Desired namespace of the pod. Default namespace is ``default``.
+    :type namespace: str
+
     :param kube_url: URL of a proxy to kubernetes cluster api. This is useful to offload authentication/authorization
                      to proxy service instead of depending on serviceaccount config. Default is ``None``.
     :type kube_url: str
 
-    :param namespace: Desired namespace of the pods. Default namespace is ``default``.
-    :type namespace: str
-
-    :return: List of pods.
-    :rtype: list
+    :return: The matching pod.
+    :rtype: pykube.Pod
     """
-    if kube_url:
-        r = requests.get(urljoin(kube_url, PODS_URL.format(namespace)))
-
-        r.raise_for_status()
-
-        return r.json().get('items', [])
-
-    kube_client = get_client()
-    return pykube.Pod.objects(kube_client).filter(namespace=namespace)
+    try:
+        if kube_url:
+            r = requests.get(urljoin(kube_url, PODS_URL.format(namespace, name)))
 
+            r.raise_for_status()
 
-def get_pod_labels_annotations(pods: list, pod_name: str) -> Tuple[dict, dict]:
-    for pod in pods:
-        metadata = pod.obj['metadata'] if hasattr(pod, 'obj') else pod.get('metadata', {})
-        if metadata.get('name') == pod_name:
-            return metadata.get('labels', {}), metadata.get('annotations', {})
+            return r.json().get('items', [])[0]
 
-    return {}, {}
+        kube_client = get_client()
+        return list(
+            pykube.Pod.objects(kube_client).filter(namespace=namespace, field_selector={'metadata.name': name}))[0]
+    except Exception:
+        raise PodNotFound('Cannot find pod: {}'.format(name))
 
 
 def is_pause_container(config: dict) -> bool:
diff --git a/kube_log_watcher/main.py b/kube_log_watcher/main.py
index d13ed51..16f6fd9 100644
--- a/kube_log_watcher/main.py
+++ b/kube_log_watcher/main.py
@@ -19,6 +19,9 @@ DEST_PATH = '/mnt/jobs/'
 APP_LABEL = 'application'
 VERSION_LABEL = 'version'
 
+ANNOTATION_PREFIX = 'annotation.'
+KUBERNETES_PREFIX = 'io.kubernetes.'
+
 BUILTIN_AGENTS = {
     'appdynamics': AppDynamicsAgent,
     'scalyr': ScalyrAgent,
@@ -32,7 +35,7 @@ logger.addHandler(logging.StreamHandler(stream=sys.stdout))
 logger.setLevel(logging.INFO)
 
 
-def get_label_value(config, label) -> str:
+def get_container_label_value(config, label) -> str:
     """
     Get label value from container config. Usually those labels are namespaced in the form:
         io.kubernetes.container.name
@@ -188,10 +191,6 @@ def get_new_containers_log_targets(
     :return: List of existing container log targets.
     :rtype: list
     """
-    pod_map = {}
-
-    pod_map[kube.DEFAULT_NAMESPACE] = kube.get_pods(kube_url=kube_url)
-
     containers_log_targets = []
 
     for container in containers:
@@ -200,21 +199,20 @@ def get_new_containers_log_targets(
 
             if kube.is_pause_container(config['Config']):
                 # We have no interest in Pause containers.
-                logger.debug('Skipping pause container({})'.format(container['id']))
                 continue
 
-            pod_name = get_label_value(config, 'pod.name')
-            container_name = get_label_value(config, 'container.name')
-            pod_namespace = get_label_value(config, 'pod.namespace')
+            pod_name = get_container_label_value(config, 'pod.name')
+            container_name = get_container_label_value(config, 'container.name')
+            pod_namespace = get_container_label_value(config, 'pod.namespace')
 
-            pods = pod_map.get(pod_namespace)
-            if not pods:
-                # We need to get pods in different namespace
-                logger.debug('Retrieving pods in namespace: {}'.format(pod_namespace))
-                pods = kube.get_pods(kube_url=kube_url, namespace=pod_namespace)
-                pod_map[pod_namespace] = pods
+            try:
+                pod = kube.get_pod(pod_name, namespace=pod_namespace, kube_url=kube_url)
+            except kube.PodNotFound:
+                logger.warning('Cannot find pod "{}" ... skipping container: {}'.format(pod_name, container_name))
+                continue
 
-            pod_labels, pod_annotations = kube.get_pod_labels_annotations(pods, pod_name)
+            metadata = pod.obj['metadata']
+            pod_labels, pod_annotations = metadata.get('labels', {}), metadata.get('annotations', {})
 
             kwargs = {}
 
@@ -243,6 +241,8 @@ def get_new_containers_log_targets(
                     continue
                 else:
                     if not kwargs['application_id']:
+                        logger.warning('Cannot determine application_id for pod: {}. Falling back to pod name!'.format(
+                            pod_name))
                         kwargs['application_id'] = kwargs['pod_name']
 
             containers_log_targets.append({'id': container['id'], 'kwargs': kwargs, 'pod_labels': pod_labels})
@@ -367,7 +367,3 @@ def main():
     logger.info('\tInterval: {}'.format(interval))
 
     watch(containers_path, agents, cluster_id, interval=interval, kube_url=kube_url, strict_labels=strict_labels)
-
-
-if __name__ == '__main__':
-    main()


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 6d5ad6a..d4bd81f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -51,7 +51,10 @@ TARGET_INVALID_ANNOT['kwargs']['pod_annotations'] = {SCALYR_ANNOTATION_PARSER: '
             {
                 'config': {
                     'Config': {
-                        'Labels': {'pod.name': 'pod-1', 'pod.namespace': 'default', 'container.name': 'cont-1'},
+                        'Labels': {
+                            'io.kubernetes.pod.name': 'pod-1', 'pod.namespace': 'default', 'container.name': 'cont-1',
+                            'annotation.some-annotation': 'v1',
+                        },
                         'Image': 'repo/example.org/cont-1:1.1'
                     }
                 },
@@ -151,7 +154,7 @@ TARGET_INVALID_ANNOT['kwargs']['pod_annotations'] = {SCALYR_ANNOTATION_PARSER: '
                     'log_file_path': '/mnt/containers/cont-5/cont-5-json.log', 'container_name': 'cont-5',
                     'pod_annotations': {}, 'image': 'cont-5', 'image_version': 'latest'
                 }
-            }
+            },
         ],
         # 4. targets no labels
         [
diff --git a/tests/test_kube.py b/tests/test_kube.py
index 9d64467..3375e8c 100644
--- a/tests/test_kube.py
+++ b/tests/test_kube.py
@@ -2,7 +2,7 @@ import pytest
 
 from mock import MagicMock
 
-from kube_log_watcher.kube import get_pods, is_pause_container, get_pod_labels_annotations, get_client
+from kube_log_watcher.kube import get_pod, is_pause_container, get_client, PodNotFound
 from kube_log_watcher.kube import PAUSE_CONTAINER_PREFIX, DEFAULT_SERVICE_ACC
 
 
@@ -52,36 +52,53 @@ def test_get_client(monkeypatch):
 
 
 @pytest.mark.parametrize('namespace', ('default', 'kube-system'))
-def test_get_pods_url(monkeypatch, namespace):
+def test_get_pod_url(monkeypatch, namespace):
     get = MagicMock()
-    res = [1, 2, 3]
+    res = [1]
     get.return_value.json.return_value = {'items': res}
 
     monkeypatch.setattr('requests.get', get)
 
-    result = get_pods(KUBE_URL, namespace=namespace)
+    result = get_pod('my-pod', namespace=namespace, kube_url=KUBE_URL)
 
-    assert res == result
+    assert res[0] == result
 
-    get.assert_called_with('https://my-kube-api/api/v1/namespaces/{}/pods'.format(namespace))
+    get.assert_called_with('https://my-kube-api/api/v1/namespaces/{}/pods/my-pod'.format(namespace))
 
 
 @pytest.mark.parametrize('namespace', ('default', 'kube-system'))
-def test_get_pods_pykube(monkeypatch, namespace):
+def test_get_pod_pykube(monkeypatch, namespace):
     get = MagicMock()
     pod = MagicMock()
-    res = [1, 2, 3]
+    res = [1]
     pod.objects.return_value.filter.return_value = res
 
     monkeypatch.setattr('kube_log_watcher.kube.get_client', get)
     monkeypatch.setattr('pykube.Pod', pod)
 
-    result = get_pods(namespace=namespace)
+    result = get_pod('my-pod', namespace=namespace)
 
-    assert res == result
+    assert res[0] == result
 
     get.assert_called_once()
-    pod.objects.return_value.filter.assert_called_with(namespace=namespace)
+    pod.objects.return_value.filter.assert_called_with(namespace=namespace, field_selector={'metadata.name': 'my-pod'})
+
+
+@pytest.mark.parametrize('namespace', ('default', 'kube-system'))
+def test_get_pod_pykube_not_found(monkeypatch, namespace):
+    get = MagicMock()
+    pod = MagicMock()
+    res = []
+    pod.objects.return_value.filter.return_value = res
+
+    monkeypatch.setattr('kube_log_watcher.kube.get_client', get)
+    monkeypatch.setattr('pykube.Pod', pod)
+
+    with pytest.raises(PodNotFound):
+        get_pod('my-pod', namespace=namespace)
+
+    get.assert_called_once()
+    pod.objects.return_value.filter.assert_called_with(namespace=namespace, field_selector={'metadata.name': 'my-pod'})
 
 
 @pytest.mark.parametrize(
@@ -96,16 +113,3 @@ def test_get_pods_pykube(monkeypatch, namespace):
 )
 def test_pause_container(monkeypatch, config, res):
     assert res == is_pause_container(config)
-
-
-@pytest.mark.parametrize(
-    'pod_name,res',
-    (
-        ('pod-1', ({'app': 'app-1', 'version': 'v1'}, {'annotation/1': 'a1', 'annotation/2': 'a2'})),
-        ('pod-2', ({'app': 'app-2', 'version': 'v1'}, {})),
-        ('pod-3', (POD_OBJ.obj['metadata']['labels'], {'annotation/1': 'a1', 'annotation/2': 'a2'})),
-        ('pod-4', ({}, {})),
-    )
-)
-def test_get_pod_labels_annotations(monkeypatch, pod_name, res):
-    assert get_pod_labels_annotations(PODS, pod_name) == res
diff --git a/tests/test_main.py b/tests/test_main.py
index 6c0f1e3..03c424a 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -4,9 +4,10 @@ import pytest
 
 from mock import MagicMock, call
 
+from kube_log_watcher.kube import PodNotFound
 from kube_log_watcher.template_loader import load_template
 from kube_log_watcher.main import (
-    get_label_value, get_containers, sync_containers_log_agents, get_stale_containers, load_agents,
+    get_container_label_value, get_containers, sync_containers_log_agents, get_stale_containers, load_agents,
     get_new_containers_log_targets, get_container_image_parts, watch)
 
 from .conftest import CLUSTER_ID
@@ -26,6 +27,13 @@ CONTAINERS_PATH = '/mnt/containers/'
 DEST_PATH = '/mnt/jobs/'
 
 
+def pod_mock(metadata):
+    pod = MagicMock()
+    pod.obj = metadata
+
+    return pod
+
+
 @pytest.mark.parametrize('image, res', (
     ('repo/image-1:0.1', ('image-1', '0.1')),
     ('', ('', 'latest')),
@@ -51,8 +59,8 @@ def test_get_container_image_parts(monkeypatch, image, res):
         ('container.nam', None),
     )
 )
-def test_get_label_value(monkeypatch, label, val):
-    assert val == get_label_value(CONFIG, label)
+def test_get_container_label_value(monkeypatch, label, val):
+    assert val == get_container_label_value(CONFIG, label)
 
 
 @pytest.mark.parametrize(
@@ -121,10 +129,10 @@ def test_get_containers(monkeypatch, walk, config, res, exc):
 def test_sync_containers_log_agents(monkeypatch, watched_containers, fx_containers_sync):
     containers, pods, targets, _, result = fx_containers_sync
 
-    get_pods = MagicMock()
-    get_pods.return_value = pods
+    get_pod = MagicMock()
+    get_pod.side_effect = pods
 
-    monkeypatch.setattr('kube_log_watcher.kube.get_pods', get_pods)
+    monkeypatch.setattr('kube_log_watcher.kube.get_pod', get_pod)
     monkeypatch.setattr('kube_log_watcher.main.CLUSTER_NODE_NAME', 'node-1')
 
     stale_containers = watched_containers - result
@@ -159,13 +167,53 @@ def test_sync_containers_log_agents(monkeypatch, watched_containers, fx_containe
         agent2.remove_log_target.assert_has_calls(remove_calls, any_order=True)
 
 
+@pytest.mark.parametrize(
+    'watched_containers',
+    (
+        set(),
+        {'cont-5'},
+        {'cont-4'},  # stale
+    )
+)
+def test_sync_containers_log_agents_failure(monkeypatch, watched_containers, fx_containers_sync):
+    containers, pods, targets, _, result = fx_containers_sync
+
+    get_pod = MagicMock()
+    get_pod.side_effect = pods
+
+    monkeypatch.setattr('kube_log_watcher.kube.get_pod', get_pod)
+    monkeypatch.setattr('kube_log_watcher.main.CLUSTER_NODE_NAME', 'node-1')
+
+    stale_containers = watched_containers - result
+    if watched_containers:
+        result = result - watched_containers
+        targets = [t for t in targets if t['id'] not in watched_containers]
+
+    get_targets = MagicMock()
+    get_targets.return_value = targets
+    get_stale = MagicMock()
+    get_stale.return_value = stale_containers
+    monkeypatch.setattr('kube_log_watcher.main.get_new_containers_log_targets', get_targets)
+    monkeypatch.setattr('kube_log_watcher.main.get_stale_containers', get_stale)
+
+    agent1 = MagicMock()
+    agent2 = MagicMock()
+    agent1.add_log_target.side_effect, agent2.add_log_target.side_effect = Exception, RuntimeError
+    agents = [agent1, agent2]
+
+    existing, stale = sync_containers_log_agents(agents, watched_containers, containers, CONTAINERS_PATH, CLUSTER_ID)
+
+    assert existing == result
+    assert stale == stale_containers
+
+
 def test_get_new_containers_log_targets(monkeypatch, fx_containers_sync):
     containers, pods, result, _, _ = fx_containers_sync
 
-    get_pods = MagicMock()
-    get_pods.return_value = pods
+    get_pod = MagicMock()
+    get_pod.side_effect = [pod_mock(p) for p in pods]
 
-    monkeypatch.setattr('kube_log_watcher.kube.get_pods', get_pods)
+    monkeypatch.setattr('kube_log_watcher.kube.get_pod', get_pod)
     monkeypatch.setattr('kube_log_watcher.main.CLUSTER_NODE_NAME', 'node-1')
 
     targets = get_new_containers_log_targets(containers, CONTAINERS_PATH, CLUSTER_ID, strict_labels=True)
@@ -173,15 +221,44 @@ def test_get_new_containers_log_targets(monkeypatch, fx_containers_sync):
     assert targets == result
 
 
+def test_get_new_containers_log_targets_not_found_pods(monkeypatch, fx_containers_sync):
+    containers, pods, _, _, _ = fx_containers_sync
+
+    get_pod = MagicMock()
+    get_pod.side_effect = PodNotFound
+
+    monkeypatch.setattr('kube_log_watcher.kube.get_pod', get_pod)
+    monkeypatch.setattr('kube_log_watcher.main.CLUSTER_NODE_NAME', 'node-1')
+
+    targets = get_new_containers_log_targets(containers, CONTAINERS_PATH, CLUSTER_ID, strict_labels=True)
+
+    assert targets == []
+
+
+def test_get_new_containers_log_targets_failuer(monkeypatch, fx_containers_sync):
+    containers, pods, _, _, _ = fx_containers_sync
+
+    is_pause = MagicMock()
+    is_pause.side_effect = Exception
+
+    # Force exception
+    monkeypatch.setattr('kube_log_watcher.kube.is_pause_container', is_pause)
+    monkeypatch.setattr('kube_log_watcher.main.CLUSTER_NODE_NAME', 'node-1')
+
+    targets = get_new_containers_log_targets(containers, CONTAINERS_PATH, CLUSTER_ID, strict_labels=True)
+
+    assert targets == []
+
+
 def test_get_new_containers_log_targets_no_strict_labels(monkeypatch, fx_containers_sync):
     containers, pods, result_labels, result_no_labels, _ = fx_containers_sync
 
     result = result_labels + result_no_labels
 
-    get_pods = MagicMock()
-    get_pods.return_value = pods
+    get_pod = MagicMock()
+    get_pod.side_effect = [pod_mock(p) for p in pods]
 
-    monkeypatch.setattr('kube_log_watcher.kube.get_pods', get_pods)
+    monkeypatch.setattr('kube_log_watcher.kube.get_pod', get_pod)
     monkeypatch.setattr('kube_log_watcher.main.CLUSTER_NODE_NAME', 'node-1')
 
     targets = get_new_containers_log_targets(containers, CONTAINERS_PATH, CLUSTER_ID)