DTIN-3938: Accommodate busy k8s clusters with short-lived pods (#1263)

Co-authored-by: ales.novak <[email protected]>
Co-authored-by: alesnovak-s1 <[email protected]>

Author: jmakar-s1

scalyr_agent/builtin_monitors/kubernetes_monitor.py

@@ -1,4 +1,4 @@
-# Copyright 2018 Scalyr Inc.
+# Copyright 2018-2024 Scalyr Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -11,8 +11,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-# ------------------------------------------------------------------------
-# author:  Imron Alston <[email protected]>
 
 from __future__ import unicode_literals
 from __future__ import absolute_import
@@ -22,11 +20,8 @@
 if False:  # NOSONAR
     from typing import Dict, List, Optional
 
-__author__ = "[email protected]"
-
 import six
 
-
 import datetime
 import docker
 import fnmatch
@@ -379,6 +374,15 @@
     env_aware=True,
 )
 
+define_config_option(
+    __monitor__,
+    "k8s_cri_query_filesystem_retain_not_found",
+    "Optional (defaults to True). If True, then when in CRI mode, the monitor retains pods found in the filesystem and not found via the Kubelet API.  Has no effect unless k8s_cri_query_filesystem is true.",
+    convert_to=bool,
+    default=True,
+    env_aware=True,
+)
+
 define_config_option(
     __monitor__,
     "k8s_verify_api_queries",
@@ -2155,6 +2159,7 @@ class CRIEnumerator(ContainerEnumerator):
 
     def __init__(
         self,
+        config,
         global_config,
         agent_pod,
         k8s_api_url,
@@ -2165,6 +2170,7 @@ def __init__(
         is_sidecar_mode=False,
     ):
         """
+        @param config: Monitor configuration
         @param global_config: Global configuration
         @param agent_pod: The QualfiedName of the agent pod.
         @param k8s_api_url: The URL to use for accessing the API
@@ -2185,7 +2191,9 @@ def __init__(
             verify_https=global_config.k8s_verify_kubelet_queries,
             ca_file=global_config.k8s_kubelet_ca_cert,
         )
+
         self._query_filesystem = query_filesystem
+        self._query_filesystem_retain_not_found = config.get("k8s_cri_query_filesystem_retain_not_found")
 
         self._log_base = "/var/log/containers"
         self._pod_base = "/var/log/pods"
@@ -2299,11 +2307,14 @@ def _get_containers(
                             ignore_k8s_api_exception=False,
                         )
                     except k8s_utils.K8sApiException as e:
-                        # If the pod details cannot be retrieved from the K8s API:
-                        # 404 - log warning, include based on k8s_include_by_default
-                        # Otherwise (401 as per K8s Api documentation) - log error
-                        # Exclude the pod
+                        # If the pod details cannot be retrieved from the api
+                        # - Due to a 404 then conditionally include and log behavior
+                        # - Due to another status code then log the error
                         if e.status_code == 404:
+                            # Optionally do not include when querying the container list by filesystem;
+                            # dead containers may not be cleaned up for an arbitrary amount of time
+                            if not self._query_filesystem_retain_not_found:
+                                continue
                             if k8s_include_by_default:
                                 include_message = "Including pod based on SCALYR_K8S_INCLUDE_ALL_CONTAINERS=true."
                             else:
@@ -2876,6 +2887,7 @@ def start(self):
                     % six.text_type(query_fs)
                 )
                 self._container_enumerator = CRIEnumerator(
+                    self._config,
                     self._global_config,
                     self.__agent_pod,
                     k8s_api_url,

scalyr_agent/configuration.py

@@ -1,4 +1,4 @@
-# Copyright 2014 Scalyr Inc.
+# Copyright 2014-2024 Scalyr Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -11,16 +11,10 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-# ------------------------------------------------------------------------
-#
-#
-# author: Steven Czerwinski <[email protected]>
 
 from __future__ import unicode_literals
 from __future__ import absolute_import
 
-__author__ = "[email protected]"
-
 from scalyr_agent.scalyr_monitor import MonitorInformation
 
 if False:
@@ -80,6 +74,7 @@
 
 DEFAULT_WORKER_ID = "default"
 
+
 def ensure_https_url(server):
     parts = six.moves.urllib.parse.urlparse(server)
 
@@ -93,6 +88,7 @@ def ensure_https_url(server):
     else:
         return server
 
+
 class Configuration(object):
     """Encapsulates the results of a single read of the configuration file.
 
@@ -155,7 +151,6 @@ def __init__(
         # FIX THESE:
         # Add documentation, verify, etc.
         self.max_retry_time = 15 * 60
-        self.max_allowed_checkpoint_age = 15 * 60
 
         # An additional directory to look for config snippets
         self.__extra_config_directory = extra_config_dir
@@ -568,7 +563,6 @@ def __verify_and_match_workers_in_logs(self):
         for worker_config in self.__config.get_json_array("workers"):
             worker_ids.add(worker_config["id"])
 
-
         # get all lists where log files entries may be defined and require worker_id param.
         # __k8s_log_configs is left out because it interferes with the kubernetes monitor container checker and CopyingManager itself sets default worker_id while adding logs.
         log_config_lists_worker_id_required = [
@@ -2091,6 +2085,10 @@ def intermediate_certs_path(self):
             "certs", "intermediate_certs.pem"
         )
 
+    @property
+    def max_allowed_checkpoint_age(self):
+        return self.__get_config().get_int("max_allowed_checkpoint_age")
+
     @staticmethod
     def __resolve_to_install_location(*paths):
         """Returns the absolute path created by joining the specified intermediate paths to
@@ -3598,6 +3596,14 @@ def __verify_main_config_and_apply_defaults(
                     error_code="invalidValue",
                 )
 
+        self.__verify_or_set_optional_int(
+            config,
+            "max_allowed_checkpoint_age",
+            15 * 60,
+            description,
+            apply_defaults,
+        )
+
     def __verify_compression_type(self, compression_type):
         """
         Verify that the library for the specified compression type (algorithm) is available.

scalyr_agent/monitor_utils/k8s.py

@@ -581,6 +581,8 @@ def __repr__(self):
         )
 
 
+# TODO Add a limit to the size (and convert to a LRU cache).
+#      This is important for caching pod info on a busy cluster.
 class _K8sCache(object):
     """
     A cached store of objects from a k8s api query

tests/unit/builtin_monitors/kubernetes_monitor_test.py

@@ -1,4 +1,4 @@
-# Copyright 2019 Scalyr Inc.
+# Copyright 2019-2024 Scalyr Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -11,9 +11,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-# ------------------------------------------------------------------------
-#
-# author: Edward Chee <[email protected]>
 
 from __future__ import unicode_literals
 from __future__ import absolute_import
@@ -51,9 +48,6 @@
 )
 from six import StringIO
 
-__author__ = "[email protected]"
-
-
 from scalyr_agent.builtin_monitors.kubernetes_monitor import (
     KubernetesMonitor,
     ControlledCacheWarmer,
@@ -1216,8 +1210,8 @@ def test_get_container_k8s_api_invalid_status(self, logger):
         """
         Mocking based test case which verifies that CRIEnumerator._get_container() correctly
         Handles the case when the Kubernetes API server returns an invalid status code.
-        404 => Pod not found, Log Warning, excluded
-        401 => Unauthorized, Log Error, excluded
+        404 => Not found, pod excluded
+        401 => Unauthorized, log error, pod excluded
         """
         self._write_file_with_separator_conversion(
             """ {
@@ -1254,6 +1248,7 @@ def mock_get_containers_from_filesystem(k8s_namespaces_to_include=None):
         ]
 
         cri = CRIEnumerator(
+            {"k8s_cri_query_filesystem_retain_not_found":True},
             global_config=global_config,
             agent_pod=mock.Mock,
             k8s_api_url="mock",
@@ -1385,6 +1380,7 @@ def mock_k8s_cache_pod(
         k8s_cache.pod.side_effect = mock_k8s_cache_pod
 
         cri = CRIEnumerator(
+            {"k8s_cri_query_filesystem_retain_not_found":True},
             global_config=global_config,
             agent_pod=mock.Mock,
             k8s_api_url="mock",

tests/unit/configuration_k8s_test.py

@@ -96,6 +96,7 @@ def test_environment_aware_module_params(self, mock_docker):
             "k8s_parse_format": (STANDARD_PREFIX, TEST_PARSE_FORMAT, six.text_type),
             "k8s_always_use_cri": (STANDARD_PREFIX, True, bool),
             "k8s_cri_query_filesystem": (STANDARD_PREFIX, True, bool),
+            "k8s_cri_query_filesystem_retain_not_found": (STANDARD_PREFIX, True, bool),
             "k8s_always_use_docker": (STANDARD_PREFIX, True, bool),
             "k8s_kubelet_host_ip": (STANDARD_PREFIX, False, bool),
             "k8s_kubelet_api_url_template": (STANDARD_PREFIX, False, bool),