Adding containerd compatability to oom_logger

paasta_tools/oom_logger.py

@@ -33,10 +33,15 @@
   destination(paasta_oom_logger);
 };
 """
+import argparse
+import json
 import re
 import sys
 from collections import namedtuple
 
+import grpc
+from containerd.services.containers.v1 import containers_pb2
+from containerd.services.containers.v1 import containers_pb2_grpc
 from docker.errors import APIError
 
 from paasta_tools.cli.utils import get_instance_config
@@ -76,6 +81,16 @@
 )
 
 
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="paasta_oom_logger")
+    parser.add_argument(
+        "--containerd",
+        action="store_true",
+        help="Use containerd to inspect containers",
+    )
+    return parser.parse_args()
+
+
 def capture_oom_events_from_stdin():
     process_name_regex = re.compile(
         r"^\d+\s[a-zA-Z0-9\-]+\s.*\]\s(.+)\sinvoked\soom-killer:"
@@ -136,11 +151,15 @@ def capture_oom_events_from_stdin():
                 break
 
 
-def get_container_env_as_dict(docker_inspect):
+def get_container_env_as_dict(is_cri_containerd: bool, container_inspect):
     env_vars = {}
-    config = docker_inspect.get("Config")
-    if config is not None:
+    if is_cri_containerd:
+        config = container_inspect.get("process")
+        env = config.get("env", [])
+    else:
+        config = container_inspect.get("Config")
         env = config.get("Env", [])
+    if config is not None:
         for i in env:
             name, _, value = i.partition("=")
             env_vars[name] = value
@@ -209,18 +228,33 @@ def send_sfx_event(service, instance, cluster):
         counter.count()
 
 
+def get_containerd_container(
+    is_cri_containerd: bool, container_id: str
+) -> containers_pb2.Container:
+    with grpc.insecure_channel("unix:///run/containerd/containerd.sock") as channel:
+        containersv1 = containers_pb2_grpc.ContainersStub(channel)
+        if is_cri_containerd:
+            namespace = "k8s.io"
+        else:
+            namespace = "moby"
+        return containersv1.Get(
+            containers_pb2.GetContainerRequest(id=container_id),
+            metadata=(("containerd-namespace", namespace),),
+        ).container
+
+
 def main():
     if clog is None:
         print("CLog logger unavailable, exiting.", file=sys.stderr)
         sys.exit(1)
-
+    args = parse_args()
     clog.config.configure(
         scribe_host="169.254.255.254",
         scribe_port=1463,
         monk_disable=False,
         scribe_disable=False,
     )
-
+    # add arg parse and parse the argument that represents a flag for containerd
     cluster = load_system_paasta_config().get_cluster()
     client = get_docker_client()
     for (
@@ -229,11 +263,18 @@ def main():
         container_id,
         process_name,
     ) in capture_oom_events_from_stdin():
-        try:
-            docker_inspect = client.inspect_container(resource_id=container_id)
-        except (APIError):
-            continue
-        env_vars = get_container_env_as_dict(docker_inspect)
+        if args.containerd:
+            # then we're using containerd to inspect containers
+            container_info = get_containerd_container(args.containerd, container_id)
+            container_spec_raw = container_info.spec.value.decode("utf-8")
+            container_inspect = json.loads(container_spec_raw)
+        else:
+            # we're using docker to inspect containers
+            try:
+                container_inspect = client.inspect_container(resource_id=container_id)
+            except (APIError):
+                continue
+        env_vars = get_container_env_as_dict(args.containerd, container_inspect)
         service = env_vars.get("PAASTA_SERVICE", "unknown")
         instance = env_vars.get("PAASTA_INSTANCE", "unknown")
         mesos_container_id = env_vars.get("MESOS_CONTAINER_NAME", "mesos-null")

requirements-minimal.txt

@@ -8,12 +8,14 @@ botocore
 bravado >= 10.2.0
 certifi
 choice >= 0.1
+containerd
 cookiecutter >= 1.4.0
 croniter
 docker
 dulwich >= 0.17.3
 ephemeral-port-reserve >= 1.0.1
 graphviz
+grpcio
 gunicorn
 humanfriendly
 humanize >= 0.5.1

requirements.txt

@@ -18,6 +18,7 @@ certifi==2017.11.5
 chardet==3.0.4
 choice==0.1
 click==6.6
+containerd==1.5.3
 cookiecutter==1.4.0
 croniter==1.3.4
 decorator==4.1.2
@@ -27,6 +28,7 @@ ephemeral-port-reserve==1.1.0
 future==0.16.0
 google-auth==1.2.0
 graphviz==0.8.2
+grpcio==1.62.2
 gunicorn==19.8.1
 http-parser==0.9.0
 humanfriendly==4.18

tests/test_oom_logger.py

@@ -21,6 +21,7 @@
 from paasta_tools.oom_logger import log_to_clog
 from paasta_tools.oom_logger import LogLine
 from paasta_tools.oom_logger import main
+from paasta_tools.oom_logger import parse_args
 from paasta_tools.oom_logger import send_sfx_event
 
 
@@ -344,14 +345,21 @@ def test_send_sfx_event(mock_get_instance_config):
         assert mock_meteorite.create_counter.return_value.count.call_count == 1
 
 
+@patch("paasta_tools.oom_logger.argparse", autospec=True)
+def test_parse_args(mock_argparse):
+    assert parse_args() == mock_argparse.ArgumentParser.return_value.parse_args()
+
+
 @patch("paasta_tools.oom_logger.sys.stdin", autospec=True)
 @patch("paasta_tools.oom_logger.clog", autospec=True)
 @patch("paasta_tools.oom_logger.send_sfx_event", autospec=True)
 @patch("paasta_tools.oom_logger.load_system_paasta_config", autospec=True)
 @patch("paasta_tools.oom_logger.log_to_clog", autospec=True)
 @patch("paasta_tools.oom_logger.log_to_paasta", autospec=True)
 @patch("paasta_tools.oom_logger.get_docker_client", autospec=True)
+@patch("paasta_tools.oom_logger.parse_args", autospec=True)
 def test_main(
+    mock_parse_args,
     mock_get_docker_client,
     mock_log_to_paasta,
     mock_log_to_clog,
@@ -365,6 +373,7 @@ def test_main(
 ):
 
     mock_sys_stdin.readline.side_effect = sys_stdin
+    mock_parse_args.return_value.containerd = False
     docker_client = Mock(inspect_container=Mock(return_value=docker_inspect))
     mock_get_docker_client.return_value = docker_client
     mock_load_system_paasta_config.return_value.get_cluster.return_value = (