text: # Copyright 2018 Scalyr Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #   http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # 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 <imron@scalyr.com>
  from __future__ import unicode_literals
  from __future__ import absolute_import
  
  __author__ = "imron@scalyr.com"
  
  from scalyr_agent import compat
  
  from scalyr_agent.monitor_utils.k8s import (
      KubernetesApi,
      K8sApiException,
      K8sApiAuthorizationException,
      ApiQueryOptions,
  )
  import scalyr_agent.monitor_utils.k8s as k8s_utils
  
  from scalyr_agent.third_party.requests.exceptions import ConnectionError
  
  import datetime
  import re
  import traceback
  import time
  import logging
  import logging.handlers
  
  import six
  import six.moves.urllib.request
  import six.moves.urllib.error
  import six.moves.urllib.parse
  
  
  from scalyr_agent import (
      ScalyrMonitor,
      define_config_option,
      AutoFlushingRotatingFileHandler,
  )
  from scalyr_agent.scalyr_logging import BaseFormatter
  from scalyr_agent.json_lib.objects import ArrayOfStrings
  import scalyr_agent.util as scalyr_util
  from scalyr_agent.json_lib import JsonObject
  
  import scalyr_agent.scalyr_logging as scalyr_logging
  
  global_log = scalyr_logging.getLogger(__name__)
  
  SCALYR_CONFIG_ANNOTATION_RE = re.compile(r"^(agent\.config\.scalyr\.com/)(.+)")
  
  __monitor__ = __name__
  
  define_config_option(
      __monitor__,
      "module",
      "Always ``scalyr_agent.builtin_monitors.kubernetes_events_monitor``",
      convert_to=six.text_type,
      required_option=True,
  )
  
  define_config_option(
      __monitor__,
      "max_log_size",
      "Optional (defaults to None). How large the log file will grow before it is rotated. If None, then the "
      "default value will be taken from the monitor level or the global level log_rotation_max_bytes config option.  Set to zero "
      "for infinite size. Note that rotation is not visible in Scalyr; it is only relevant for managing "
      "disk space on the host running the agent. However, a very small limit could cause logs to be "
      "dropped if there is a temporary network outage and the log overflows before it can be sent to "
      "Scalyr",
      convert_to=int,
      default=None,
      env_name="SCALYR_K8S_MAX_LOG_SIZE",
  )
  
  define_config_option(
      __monitor__,
      "max_log_rotations",
      "Optional (defaults to None). The maximum number of log rotations before older log files are "
      "deleted. If None, then the value is taken from the monitor level or the global level log_rotation_backup_count option. "
      "Set to zero for infinite rotations.",
      convert_to=int,
      default=None,
      env_name="SCALYR_K8S_MAX_LOG_ROTATIONS",
  )
  
  define_config_option(
      __monitor__,
      "log_flush_delay",
      "Optional (defaults to 1.0). The time to wait in seconds between flushing the log file containing "
      "the kubernetes event messages.",
      convert_to=float,
      default=1.0,
      env_name="SCALYR_K8S_LOG_FLUSH_DELAY",
  )
  
  define_config_option(
      __monitor__,
      "message_log",
      "Optional (defaults to ``kubernetes_events.log``). Specifies the file name under which event messages "
      "are stored. The file will be placed in the default Scalyr log directory, unless it is an "
      "absolute path",
      convert_to=six.text_type,
      default="kubernetes_events.log",
      env_name="SCALYR_K8S_MESSAGE_LOG",
  )
  
  EVENT_OBJECT_FILTER_DEFAULTS = [
      "CronJob",
      "DaemonSet",
      "Deployment",
      "Job",
      "Node",
      "Pod",
      "ReplicaSet",
      "ReplicationController",
      "StatefulSet",
  ]
  define_config_option(
      __monitor__,
      "event_object_filter",
      "Optional (defaults to %s). A list of event object types to filter on. "
      "If set, only events whose `involvedObject` `kind` is on this list will be included."
      % six.text_type(EVENT_OBJECT_FILTER_DEFAULTS),
      convert_to=ArrayOfStrings,
      default=EVENT_OBJECT_FILTER_DEFAULTS,
      env_name="SCALYR_K8S_EVENT_OBJECT_FILTER",
  )
  
  define_config_option(
      __monitor__,
      "leader_check_interval",
      "Optional (defaults to 60). The number of seconds to wait between checks to see if we are still the leader.",
      convert_to=int,
      default=60,
      env_name="SCALYR_K8S_LEADER_CHECK_INTERVAL",
  )
  
  define_config_option(
      __monitor__,
      "leader_node",
      "Optional (defaults to None). Force the `leader` to be the scalyr-agent that runs on this node.",
      convert_to=six.text_type,
      default=None,
      env_name="SCALYR_K8S_LEADER_NODE",
  )
  
  define_config_option(
      __monitor__,
      "check_labels",
      "Optional (defaults to False). If true, then the monitor will check for any nodes with the label "
      "`agent.config.scalyr.com/events_leader_candidate=true` and the node with this label set and that has the oldest"
      "creation time will be the event monitor leader.",
      convert_to=bool,
      default=False,
      env_name="SCALYR_K8S_CHECK_LABELS",
  )
  
  define_config_option(
      __monitor__,
      "ignore_master",
      "Optional (defaults to True). If true, then the monitor will ignore any nodes with the label "
      "`node-role.kubernetes.io/master` when determining which node is the event monitor leader.",
      convert_to=bool,
      default=True,
      env_name="SCALYR_K8S_IGNORE_MASTER",
  )
  
  
  class EventLogFormatter(BaseFormatter):
      """Formatter used for the logs produced by the event monitor.
  
      """
  
      def __init__(self):
          # TODO: It seems on Python 2.4 the filename and line number do not work correctly.  I think we need to
          # define a custom findCaller method to actually fix the problem.
          BaseFormatter.__init__(
              self, "%(asctime)s [kubernetes_events_monitor] %(message)s", "k8s_events"
          )
  
  
  class KubernetesEventsMonitor(ScalyrMonitor):
      """
      # Kubernetes Events Monitor
  
      The Kuberntes Events monitor streams Kubernetes events from the Kubernetes API
  
      This monitor fetches Kubernetes Events from the Kubernetes API server and uploads them to Scalyr. [Kubernetes Events](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.10/#event-v1-core) record actions taken by the Kubernetes master, such as starting or killing pods. This information can be extremely useful in debugging issues with your Pods and investigating the general health of your K8s cluster. By default, this monitor collects all events related to Pods (including objects that control Pods such as DaemonSets and Deployments) as well as Nodes.
  
      Each Event is uploaded to Scalyr as a single log line. The format is described in detail below, but in general, it has two JSON objects, one holding the contents of the Event as described by the Kubernetes API and the second holding additional information the Scalyr Agent has added to provide more context for the Event, such as the Deployment to which the Pod belongs.
  
      This monitor powers the [Kubernetes Events dashboard](https://www.scalyr.com/dash?page=k8s%20Events). Additionally, you can see all Events uploaded by [searching for monitor="kubernetes_events_monitor"](https://www.scalyr.com/events?filter=monitor%3D%22kubernetes_events_monitor%22)
  
      ## Event Collection Leader
  
      The Scalyr Agent uses a simple leader election algorithm to decide which Scalyr Agent should retrieve the cluster's Events and upload them to Scalyr. This is necessary because the Events pertain to the entire cluster (and not necessarily just one Node). Duplicate information would be uploaded if each Scalyr Agent Pod retrieved the cluster's Events and uploaded them.
  
      By default, the leader election algorithm selects the Scalyr Agent Pod running on the oldest Node, excluding the Node running the master. To determine which is the oldest node, each Scalyr Agent Pod will retrieve the entire cluster Node list at start up, and then query the API Server once every `leader_check_interval` seconds (defaults to 60 seconds) to ensure the leader is still alive. For large clusters, performing these checks can place noticeable load on the K8s API server. There are two ways to address this issue, with different impacts on load and flexibility.
  
      ### Node Labels
  
      The first approach is to add the label `agent.config.scalyr.com/events_leader_candidate=true` to any node that you wish to be eligible to become the events collector.  This can be done with the following command:
  
      `kubectl label node <nodename> agent.config.scalyr.com/events_leader_candidate=true`
  
      Where `<nodename>` is the name of the node in question.  You can set this label on multiple nodes, and the agent will use the oldest of them as the event collector leader.  If the leader node is shutdown, it will fallback to the next oldest node and so on.
  
      Once the labels have been set, you also need to configure the agent to query these labels.  This is done via the `check_labels` configuration option in the Kubernetes Events monitor config (which is off by default):
  
      ```
      ...
      monitors: [
        {
          "module": "scalyr_agent.builtin_monitors.kubernetes_events_monitor",
          ...
          "check_labels": true
        }
      ]
      ...
      ```
  
      If `check_labels` is true then instead of querying the API server for all nodes on the cluster, the Scalyr agents will only query the API server for nodes with this label set.  In order to reduce load on the API server, it is recommended to only set this label on a small number of nodes.
  
      This approach reduces the load placed on the API server and also provides a convenient fallback mechanism for when an exister event collector leader node is shutdown.
  
      ### Hardcoded Event Collector
  
      The second approach is to manually assign a leader node in the agent config.  This is done via the `leader_node` option of the Kubernetes Events monitor:
  
      ```
      ...
      monitors: [
        {
          "module": "scalyr_agent.builtin_monitors.kubernetes_events_monitor",
          ...
          "leader_node": "<name of leader node>"
        }
      ]
      ...
      ```
  
      When the leader node is explicitly set like this, no API queries are made to the K8s API server and only the node with that name will be used to query K8s events.  The downside to this approach is that it is less flexible, especially in the event of the node shutting down unexpectedly.
  
      The leader election algorithm relies on a few assumptions to work correctly and could, for large clusters, impact performance. If necessary, the events monitor can also be disabled.
  
      ## Disabling the Kubernetes Events Monitor
  
      This monitor was released and enabled by default in Scalyr Agent version `2.0.43`. If you wish to disable this monitor, you may do so by setting `k8s_events_disable` to `true` in the `scalyr-config` ConfigMap used to run your ScalyrAgent DaemonSet. You must restart the ScalyrAgent DaemonSet once you have made this configuration change. Here is one way to perform these actions:
  
      1. Fetch the `scalyr-config` ConfigMap from your cluster.
  
         ```
         kubectl get configmap scalyr-config -o yaml --export >> scalyr-config.yaml
         ```
  
      2. Add the following line to the data section of scalyr-config.yaml:
  
        ```
        k8s_events_disable: true
        ```
  
      3. Update the scalyr-config ConfigMap.
  
         ```
         kubectl delete configmap scalyr-config
         kubectl create -f scalyr-config.yaml
         ```
  
      4. Delete the existing Scalyr k8s DaemonSet.
  
         ```
         kubectl delete -f https://raw.githubusercontent.com/scalyr/scalyr-agent-2/release/k8s/scalyr-agent-2.yaml
         ```
  
      5. Run the new Scalyr k8s DaemonSet.
  
         ```
         kubectl create -f https://raw.githubusercontent.com/scalyr/scalyr-agent-2/release/k8s/scalyr-agent-2.yaml
         ```
      """
  
      def _initialize(self):
  
          # our disk logger and handler
          self.__disk_logger = None
          self.__log_handler = None
  
          self.__log_watcher = None
          self.__module = None
  
          self._node_name = None
  
          # configure the logger and path
          self.__message_log = self._config.get("message_log")
  
          self.log_config = {
              "parser": "k8sEvents",
              "path": self.__message_log,
          }
  
          self._leader_check_interval = self._config.get("leader_check_interval")
          self._leader_node = self._config.get("leader_node")
          self._check_labels = self._config.get("check_labels")
          self._ignore_master = self._config.get("ignore_master")
  
          self._current_leader = self._leader_node
  
          self.__flush_delay = self._config.get("log_flush_delay")
          try:
              attributes = JsonObject({"monitor": "json"})
              self.log_config["attributes"] = attributes
          except Exception:
              global_log.error(
                  "Error setting monitor attribute in KubernetesEventMonitor"
              )
  
          # The namespace whose logs we should not collect.
          self.__k8s_namespaces_to_ignore = []
          for x in self._global_config.k8s_ignore_namespaces:
              self.__k8s_namespaces_to_ignore.append(x.strip())
  
          (
              default_rotation_count,
              default_max_bytes,
          ) = self._get_log_rotation_configuration()
  
          self.__event_object_filter = self._config.get(
              "event_object_filter",
              [
                  "CronJob",
                  "DaemonSet",
                  "Deployment",
                  "Job",
                  "Node",
                  "Pod",
                  "ReplicaSet",
                  "ReplicationController",
                  "StatefulSet",
              ],
          )
  
          self.__max_log_size = self._config.get("max_log_size")
          if self.__max_log_size is None:
              self.__max_log_size = default_max_bytes
  
          self.__max_log_rotations = self._config.get("max_log_rotations")
          if self.__max_log_rotations is None:
              self.__max_log_rotations = default_rotation_count
  
          # Support legacy disabling of k8s_events via the K8S_EVENTS_DISABLE environment variable
          k8s_events_disable_envar = compat.os_environ_unicode.get("K8S_EVENTS_DISABLE")
          if k8s_events_disable_envar is not None:
              global_log.warn(
                  "The K8S_EVENTS_DISABLE environment variable is deprecated. Please use SCALYR_K8S_EVENTS_DISABLE instead."
              )
              legacy_disable = six.text_type(k8s_events_disable_envar).lower()
          else:
              legacy_disable = "false"
  
          # Note, accepting just a single `t` here due to K8s ConfigMap issues with having a value of `true`
          self.__disable_monitor = (
              legacy_disable == "true"
              or legacy_disable == "t"
              or self._global_config.k8s_events_disable
          )
  
      def open_metric_log(self):
          """Override open_metric_log to prevent a metric log from being created for the Kubernetes Events Monitor
          and instead create our own logger which will log raw messages out to disk.
          """
          self.__disk_logger = logging.getLogger(self.__message_log)
  
          # assume successful for when the logger handler has already been created
          success = True
  
          # only configure once -- assumes all configuration happens on the same thread
          if len(self.__disk_logger.handlers) == 0:
              # logger handler hasn't been created yet, so assume unsuccssful
              success = False
              try:
                  self.__log_handler = AutoFlushingRotatingFileHandler(
                      filename=self.log_config["path"],
                      maxBytes=self.__max_log_size,
                      backupCount=self.__max_log_rotations,
                      flushDelay=self.__flush_delay,
                  )
  
                  self.__log_handler.setFormatter(EventLogFormatter())
                  self.__disk_logger.addHandler(self.__log_handler)
                  self.__disk_logger.setLevel(logging.INFO)
                  self.__disk_logger.propagate = False
                  success = True
              except Exception as e:
                  global_log.error(
                      "Unable to open KubernetesEventsMonitor log file: %s"
                      % six.text_type(e)
                  )
  
          return success
  
      def close_metric_log(self):
          if self.__log_handler:
              self.__disk_logger.removeHandler(self.__log_handler)
              self.__log_handler.close()
  
      def set_log_watcher(self, log_watcher):
          self.__log_watcher = log_watcher
  
      def _check_if_alive(self, k8s, node):
          """
              Checks to see if the node specified by `node` is alive
          """
          if node is None:
              return False
  
          try:
              # this call will throw an exception on failure
              k8s.query_api_with_retries(
                  "/api/v1/nodes/%s" % node,
                  retry_error_context=node,
                  retry_error_limit_key="k8se_check_if_alive",
              )
          except Exception:
              global_log.log(
                  scalyr_logging.DEBUG_LEVEL_1, "_check_if_alive False for node %s" % node
              )
              return False
  
          # if we are here, then the above node exists so return True
          return True
  
      def _get_oldest_node(self, nodes, ignore_master):
          """
              Takes a list of nodes returned from querying the k8s api, and returns the name
              of the node with the oldest creationTimestamp or None if nodes is empty or doesn't
              contain Node objects.
  
              if `ignore_master` is true, then any nodes with the label `node-role.kubernetes.io/master` are ignored.
          """
  
          oldest_time = datetime.datetime.utcnow()
          oldest = None
  
          # loop over all nodes and find the one with the oldest create time
          for node in nodes:
              metadata = node.get("metadata", {})
              create_time = metadata.get("creationTimestamp", None)
              if create_time is None:
                  continue
  
              name = metadata.get("name", None)
              if name is None:
                  continue
  
              # skip the master node if necessary
              labels = metadata.get("labels", {})
              if ignore_master and "node-role.kubernetes.io/master" in labels:
                  continue
  
              # convert to datetime
              create_time = scalyr_util.rfc3339_to_datetime(create_time)
  
              # if we are older than the previous oldest datetime, then update the oldest time
              if create_time is not None and create_time < oldest_time:
                  oldest_time = create_time
                  oldest = name
  
          return oldest
  
      def _check_nodes_for_leader(self, k8s, query_fields=""):
          """
              Checks all nodes on the cluster to see which one has the oldest creationTime
              If `self._ignore_master` is True then any node with the label `node-role.kubernetes.io/master`
              is ignored.
              @param k8s: a KubernetesApi object for querying the k8s api
              @query_fields - optional query string appended to the node endpoint to allow for filtering
          """
          response = k8s.query_api_with_retries(
              "/api/v1/nodes%s" % query_fields,
              retry_error_context="nodes%s" % query_fields,
              retry_error_limit_key="k8se_check_nodes_for_leader",
          )
          nodes = response.get("items", [])
  
          if len(nodes) > 100:
              # TODO: Add in URL for how to use labels to rectify this once we have the documentation written up.
              global_log.warning(
                  "Warning, Kubernetes Events monitor leader election is finding more than 100 possible "
                  "candidates.  This could impact performance.  Contact support@scalyr.com for more "
                  "information",
                  limit_once_per_x_secs=300,
                  limit_key="k8s-too-many-event-leaders",
              )
          # sort the list by node name to ensure consistent processing order
          # if two nodes have exactly the same creation time
          nodes.sort(key=lambda n: n.get("metadata", {}).get("name", ""))
  
          # the leader is the longest running node
          return self._get_oldest_node(nodes, self._ignore_master)
  
      def _get_current_leader(self, k8s):
          """
              Queries the kubernetes api to see which node is the current leader node.
  
              If there is currently a leader node, then only that node is polled to see if it still
              exists.
  
              Check to see if the leader is specified as a node label.
              If not, then if there is not a current leader node, or if the current leader node no longer exists
              then query all nodes for the leader
              @param k8s: a KubernetesApi object for querying the k8s api
          """
  
          try:
              new_leader = None
              # first check to see if the leader node is specified as a label
              # we do this first in case the label has been moved to a different node but the old node is still
              # alive
              if self._check_labels:
                  query = "?labelSelector=%s" % (
                      six.moves.urllib.parse.quote(
                          "agent.config.scalyr.com/events_leader_candidate=true"
                      )
                  )
                  new_leader = self._check_nodes_for_leader(k8s, query)
  
              # if no labelled leader was found, then check to see if the previous leader is still alive
              if new_leader is not None:
                  global_log.log(scalyr_logging.DEBUG_LEVEL_1, "Leader set from label")
              else:
                  if self._current_leader is not None and self._check_if_alive(
                      k8s, self._current_leader
                  ):
                      # still the same leader
                      global_log.log(
                          scalyr_logging.DEBUG_LEVEL_1,
                          "New leader same as the old leader",
                      )
                      new_leader = self._current_leader
                  else:
                      # previous leader is not alive so check all nodes for the leader
                      global_log.log(
                          scalyr_logging.DEBUG_LEVEL_1, "Checking for a new leader"
                      )
                      new_leader = self._check_nodes_for_leader(k8s)
  
              # update the global leader (leader can be None)
              self._current_leader = new_leader
  
          except K8sApiAuthorizationException:
              global_log.warning(
                  "Could not determine K8s event leader due to authorization error.  The "
                  "Scalyr Service Account does not have permission to retrieve the list of "
                  "available nodes.  Please recreate the role with the latest definition which can be found "
                  "at https://raw.githubusercontent.com/scalyr/scalyr-agent-2/release/k8s/scalyr-service-account.yaml "
                  "K8s event collection will be disabled until this is resolved.  See the K8s install "
                  "directions for instructions on how to create the role "
                  "https://www.scalyr.com/help/install-agent-kubernetes",
                  limit_once_per_x_secs=300,
                  limit_key="k8s-events-no-permission",
              )
              self._current_leader = None
              return None
          except K8sApiException as e:
              global_log.error(
                  "get current leader: %s, %s"
                  % (six.text_type(e), traceback.format_exc())
              )
  
          return self._current_leader
  
      def _is_leader(self, k8s):
          """
          Detects if this agent is the `leader` for events in a cluster.  In order to prevent duplicates, only `leader` agents will log events.
          """
          if self._node_name is None:
              return False
  
          # see if we have a fixed leader node specified
          leader = self._leader_node
  
          try:
              # if not then query the k8s api for the leader
              if leader is None:
                  leader = self._get_current_leader(k8s)
  
              # check if this node is the current leader node
          except Exception as e:
              global_log.log(
                  scalyr_logging.DEBUG_LEVEL_0,
                  "Unexpected error checking for leader: %s" % (six.text_type(e)),
              )
  
          return leader is not None and self._node_name == leader
  
      def _is_resource_expired(self, response):
          """
              Checks to see if the resource identified in `response` is expired or not.
              @param response: a response from querying the k8s api for a resource
          """
          obj = response.get("object", dict())
  
          # check to see if the resource version we are using has expired
          return (
              response.get("type", "") == "ERROR"
              and obj.get("kind", "") == "Status"
              and obj.get("reason", "") == "Expired"
          )
  
      def _get_involved_object(self, obj):
          """
              Processes an object returned from querying the k8s api, and determines whether the object
              has an `involvedObject` and if so returns a tuple containing the kind, namespace and name of the involved object.
              Otherwise returns (None, None, None)
              @param obj: an object returned from querying th k8s api
          """
          involved = obj.get("involvedObject", dict())
          kind = involved.get("kind", None)
          namespace = None
          name = None
  
          if kind:
              name = involved.get("name", None)
              namespace = involved.get("namespace", None)
  
          return (kind, namespace, name)
  
      def run(self):
          """Begins executing the monitor, writing metric output to logger.
          """
          if self.__disable_monitor:
              global_log.info(
                  "kubernetes_events_monitor exiting because it has been disabled."
              )
              return
  
          try:
              self._global_config.k8s_api_url
              self._global_config.k8s_verify_api_queries
  
              # We only create the k8s_cache while we are the leader
              k8s_cache = None
  
              if self.__log_watcher:
                  self.log_config = self.__log_watcher.add_log_config(
                      self.module_name, self.log_config
                  )
  
              # First instance of k8s api uses the main rate limiter.  Leader election related API calls to the k8s
              # masters will go through this api/rate limiter.
              k8s_api_main = KubernetesApi.create_instance(
                  self._global_config, rate_limiter_key="K8S_CACHE_MAIN_RATELIMITER"
              )
  
              # Second instance of k8s api uses an ancillary ratelimiter (for exclusive use by events monitor)
              k8s_api_events = KubernetesApi.create_instance(
                  self._global_config, rate_limiter_key="K8S_EVENTS_RATELIMITER"
              )
  
              # k8s_cache is initialized with the main rate limiter. However, streaming-related API calls should go
              # through the ancillary ratelimiter. This is achieved by passing ApiQueryOptions with desired rate_limiter.
              k8s_events_query_options = ApiQueryOptions(
                  max_retries=self._global_config.k8s_controlled_warmer_max_query_retries,
                  rate_limiter=k8s_api_events.default_query_options.rate_limiter,
              )
  
              pod_name = k8s_api_main.get_pod_name()
              self._node_name = k8s_api_main.get_node_name(pod_name)
              cluster_name = k8s_api_main.get_cluster_name()
  
              last_event = None
              last_resource = 0
  
              last_check = time.time() - self._leader_check_interval
  
              last_reported_leader = None
              while not self._is_thread_stopped():
                  current_time = time.time()
  
                  # if we are the leader, we could be going through this loop before the leader_check_interval
                  # has expired, so make sure to only check for a new leader if the interval has expired
                  if last_check + self._leader_check_interval <= current_time:
                      last_check = current_time
                      # check if we are the leader
                      if not self._is_leader(k8s_api_main):
                          # if not, then sleep and try again
                          global_log.log(
                              scalyr_logging.DEBUG_LEVEL_1,
                              "Leader is %s" % (six.text_type(self._current_leader)),
                          )
                          if (
                              self._current_leader is not None
                              and last_reported_leader != self._current_leader
                          ):
                              global_log.info(
                                  "Kubernetes event leader is %s"
                                  % six.text_type(self._current_leader)
                              )
                              last_reported_leader = self._current_leader
                          if k8s_cache is not None:
                              k8s_cache.stop()
                              k8s_cache = None
                          self._sleep_but_awaken_if_stopped(self._leader_check_interval)
                          continue
  
                      global_log.log(
                          scalyr_logging.DEBUG_LEVEL_1,
                          "Leader is %s" % (six.text_type(self._current_leader)),
                      )
                  try:
                      if last_reported_leader != self._current_leader:
                          global_log.info("Acting as Kubernetes event leader")
                          last_reported_leader = self._current_leader
  
                      if k8s_cache is None:
                          # create the k8s cache
                          k8s_cache = k8s_utils.cache(self._global_config)
  
                      # start streaming events
                      lines = k8s_api_events.stream_events(last_event=last_event)
  
                      json = {}
                      for line in lines:
                          try:
                              json = scalyr_util.json_decode(line)
                          except Exception as e:
                              global_log.warning(
                                  "Error parsing event json: %s, %s, %s"
                                  % (line, six.text_type(e), traceback.format_exc())
                              )
                              continue
  
                          try:
                              # check to see if the resource version we are using has expired
                              if self._is_resource_expired(json):
                                  last_event = None
                                  global_log.log(
                                      scalyr_logging.DEBUG_LEVEL_1, "K8S resource expired"
                                  )
                                  continue
  
                              obj = json.get("object", dict())
                              event_type = json.get("type", "UNKNOWN")
  
                              # resource version hasn't expired, so update it to the most recently seen version
                              last_event = last_resource
  
                              metadata = obj.get("metadata", dict())
  
                              # skip any events with resourceVersions higher than ones we've already seen
                              resource_version = metadata.get("resourceVersion", None)
                              if resource_version is not None:
                                  resource_version = int(resource_version)
                              if resource_version and resource_version <= last_resource:
                                  global_log.log(
                                      scalyr_logging.DEBUG_LEVEL_2,
                                      "Skipping older resource events",
                                  )
                                  continue
  
                              last_resource = resource_version
                              last_event = resource_version
  
                              # see if this event is about an object we are interested in
                              (kind, namespace, name) = self._get_involved_object(obj)
  
                              if kind is None:
                                  global_log.log(
                                      scalyr_logging.DEBUG_LEVEL_1,
                                      "Ignoring event due to None kind",
                                  )
                                  continue
  
                              # exclude any events that don't involve objects we are interested in
                              if (
                                  self.__event_object_filter
                                  and kind not in self.__event_object_filter
                              ):
                                  global_log.log(
                                      scalyr_logging.DEBUG_LEVEL_1,
                                      "Ignoring event due to unknown kind %s - %s"
                                      % (kind, six.text_type(metadata)),
                                  )
                                  continue
  
                              # ignore events that belong to namespaces we are not interested in
                              if namespace in self.__k8s_namespaces_to_ignore:
                                  global_log.log(
                                      scalyr_logging.DEBUG_LEVEL_1,
                                      "Ignoring event due to belonging to an excluded namespace '%s'"
                                      % (namespace),
                                  )
                                  continue
  
                              # get cluster and deployment information
                              extra_fields = {
                                  "k8s-cluster": cluster_name,
                                  "watchEventType": event_type,
                              }
                              if kind:
                                  if kind == "Pod":
                                      extra_fields["pod_name"] = name
                                      extra_fields["pod_namespace"] = namespace
                                      pod = k8s_cache.pod(
                                          namespace,
                                          name,
                                          current_time,
                                          query_options=k8s_events_query_options,
                                      )
                                      if pod and pod.controller:
                                          extra_fields[
                                              "k8s-controller"
                                          ] = pod.controller.name
                                          extra_fields["k8s-kind"] = pod.controller.kind
                                  elif kind != "Node":
                                      controller = k8s_cache.controller(
                                          namespace,
                                          name,
                                          kind,
                                          current_time,
                                          query_options=k8s_events_query_options,
                                      )
                                      if controller:
                                          extra_fields["k8s-controller"] = controller.name
                                          extra_fields["k8s-kind"] = controller.kind
  
                              # if so, log to disk
                              self.__disk_logger.info(
                                  "event=%s extra=%s"
                                  % (
                                      six.text_type(scalyr_util.json_encode(obj)),
                                      six.text_type(
                                          scalyr_util.json_encode(extra_fields)
                                      ),
                                  )
                              )
  
                              # see if we need to check for a new leader
                              if last_check + self._leader_check_interval <= current_time:
                                  global_log.log(
                                      scalyr_logging.DEBUG_LEVEL_1,
                                      "Time to check for a new event leader",
                                  )
                                  break
  
                          except Exception as e:
                              global_log.exception(
                                  "Failed to process single k8s event line due to following exception: %s, %s, %s"
                                  % (repr(e), six.text_type(e), traceback.format_exc()),
                                  limit_once_per_x_secs=300,
                                  limit_key="k8s-stream-events-general-exception",
                              )
                  except K8sApiAuthorizationException:
                      global_log.warning(
                          "Could not stream K8s events due to an authorization error.  The "
                          "Scalyr Service Account does not have permission to watch available events.  "
                          "Please recreate the role with the latest definition which can be found "
                          "at https://raw.githubusercontent.com/scalyr/scalyr-agent-2/release/k8s/scalyr-service-account.yaml "
                          "K8s event collection will be disabled until this is resolved.  See the K8s install "
                          "directions for instructions on how to create the role "
                          "https://www.scalyr.com/help/install-agent-kubernetes",
                          limit_once_per_x_secs=300,
                          limit_key="k8s-stream-events-no-permission",
                      )
                  except ConnectionError:
                      # ignore these, and just carry on querying in the next loop
                      pass
                  except Exception as e:
                      global_log.exception(
                          "Failed to stream k8s events due to the following exception: %s, %s, %s"
                          % (repr(e), six.text_type(e), traceback.format_exc())
                      )
  
              if k8s_cache is not None:
                  k8s_cache.stop()
                  k8s_cache = None
  
          except Exception:
              # TODO:  Maybe remove this catch here and let the higher layer catch it.  However, we do not
              # right now join on the monitor threads, so no one would catch it.  We should change that.
              global_log.exception(
                  "Monitor died due to exception:", error_code="failedMonitor"
              )
  
      def stop(self, wait_on_join=True, join_timeout=5):
          # stop the main server
          ScalyrMonitor.stop(self, wait_on_join=wait_on_join, join_timeout=join_timeout)
  