Fixing an issue with multi acount support. (#1259)

Author: alesnovak-s1

k8s/scalyr-service-account.yaml

@@ -11,7 +11,7 @@ metadata:
   name: scalyr-clusterrole
 rules:
 - apiGroups: [""]
-  resources: ["namespaces","pods","replicationcontrollers"]
+  resources: ["namespaces","pods","replicationcontrollers", "secrets"]
   verbs: ["get","list"]
 - apiGroups: [""]
   resources: ["nodes"]

scalyr_agent/builtin_monitors/kubernetes_monitor.py

@@ -3476,13 +3476,18 @@ def __get_log_config_for_container(self, cid, info, k8s_cache, base_attributes):
         # Based on the pod annotations in a format {container_name}.{team}.secret={secret_name}
         # we might want to add api_keys parameter
 
-        if container_annotations or all_annotations or namespace_annotations:
+        self._logger.log(
+            scalyr_logging.DEBUG_LEVEL_0,
+            "log_config_for_container Checking for teams in annotations for container %s(%s), pod %s, namespace %s. Container annotations = %s, Pod annotations = %s, Namespace annotations = %s  " \
+            % (info["name"], short_cid, pod_name, pod_namespace, container_annotations, all_annotations, namespace_annotations)
+        )
 
-            api_keys = self.__container_api_keys_from_annotations(k8s_cache, pod_namespace, container_annotations)
+        if container_annotations or all_annotations or namespace_annotations:
+            api_keys = self.__container_api_keys_from_annotations(k8s_cache, pod_namespace, container_annotations, "container", info["name"])
             if not api_keys:
-                api_keys = self.__container_api_keys_from_annotations(k8s_cache, pod_namespace, all_annotations)
+                api_keys = self.__container_api_keys_from_annotations(k8s_cache, pod_namespace, all_annotations, "pod", info["name"])
             if not api_keys:
-                api_keys = self.__container_api_keys_from_annotations(k8s_cache, pod_namespace, namespace_annotations)
+                api_keys = self.__container_api_keys_from_annotations(k8s_cache, pod_namespace, namespace_annotations, "namespace", info["name"])
             if api_keys:
                 # Multiple matching api keys will result in multiple log configs, which will differ in the api_key field only.
                 results = [
@@ -3492,7 +3497,7 @@ def __get_log_config_for_container(self, cid, info, k8s_cache, base_attributes):
 
         return results
 
-    def __container_api_keys_from_annotations(self, k8s_cache, namespace, annotations):
+    def __container_api_keys_from_annotations(self, k8s_cache, namespace, annotations, annotation_kind, container_name):
         def fetch_secret(name):
             if not name:
                 return None
@@ -3530,17 +3535,31 @@ def get_secret_api_key(secret):
 
             return api_key
 
-        api_keys = [
-            get_secret_api_key(fetch_secret(team.get("secret")))
+        secrets_names = [
+            team.get("secret")
             for team in annotations.get("teams", [])
         ]
 
-        return [
+        secrets = [
+            secret
+            for secret in map(fetch_secret, secrets_names)
+            if secret
+        ]
+
+        api_keys = [
             api_key
-            for api_key in api_keys
+            for api_key in map(get_secret_api_key, secrets)
             if api_key
         ]
 
+        self._logger.log(
+            scalyr_logging.DEBUG_LEVEL_0,
+            "log_config_for_container From %s annotations of container %s and namespace %s, got %d non-empty api keys, %d secrets for secret names %s" \
+            % (annotation_kind, container_name, namespace, len(api_keys), len(secrets), ",".join(secrets_names))
+        )
+
+        return api_keys
+
     def __get_docker_logs(self, containers, k8s_cache):
         # type: (Dict, KubernetesCache) -> List[DockerLog]
         """Returns a list of dicts containing the container id, stream, and a log_config

scalyr_agent/configuration.py

@@ -568,6 +568,18 @@ 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 = [
+            self.__log_configs,
+            self.__journald_log_configs,
+        ]
+
+        for log_config_list in log_config_lists_worker_id_required:
+            for log_file_config in log_config_list:
+                log_file_config["worker_id"] = log_file_config.get("worker_id", default_value=DEFAULT_WORKER_ID)
+
         # get all lists where log files entries may be defined.
         log_config_lists = [
             self.__log_configs,
@@ -578,11 +590,7 @@ def __verify_and_match_workers_in_logs(self):
         for log_config_list in log_config_lists:
             for log_file_config in log_config_list:
                 worker_id = log_file_config.get("worker_id", none_if_missing=True)
-
-                if worker_id is None:
-                    # set a default worker if worker_id is not specified.
-                    log_file_config["worker_id"] = DEFAULT_WORKER_ID
-                else:
+                if worker_id is not None:
                     # if log file entry has worker_id which is not defined in the 'workers' list, then throw an error.
                     if worker_id not in worker_ids:
                         valid_worker_ids = ", ".join(sorted(worker_ids))

scalyr_agent/copying_manager/copying_manager.py

@@ -609,11 +609,44 @@ def config(self):
     def __worker_id_from_log_config(self, log_config):
         # type: (Dict) -> six.text_type
 
+        log.log(
+            scalyr_logging.DEBUG_LEVEL_0,
+            (
+                "Finding suitable worker for path %s, log_config_keys=%s"
+                % (log_config.get("path"), log_config.keys())
+            )
+        )
+
         if "worker_id" in log_config:
+            log.info(
+                (
+                    "Using worker_id %s from log_config for path %s."
+                    % (log_config['worker_id'], log_config.get('path'))
+                )
+            )
+
             return log_config["worker_id"]
 
         if "api_key" in log_config:
-            return self.__dynamic_workers.get_or_create_worker(log_config["api_key"], self.__config).worker_id
+            worker_id = self.__dynamic_workers.get_or_create_worker(log_config["api_key"], self.__config).worker_id
+
+            log.log(
+                scalyr_logging.DEBUG_LEVEL_0,
+                (
+                    "Using dynamic worker %s based on api_key from log_config for path %s."
+                    % (worker_id, log_config.get('path'))
+                )
+            )
+
+            return worker_id
+
+        log.log(
+            scalyr_logging.DEBUG_LEVEL_0,
+            (
+                "Fallback to default worker_id %s for path %s."
+                % (DEFAULT_WORKER_ID, log_config.get('path'))
+            )
+        )
 
         return DEFAULT_WORKER_ID
 
@@ -628,6 +661,7 @@ def add_log_config(self, monitor_name, log_config, force_add=False):
         the container restart where the log file is not immediately removed.
         returns: an updated log_config object
         """
+
         worker_id = self.__worker_id_from_log_config(log_config)
         log_config["worker_id"] = worker_id
         log_config = self.__config.parse_log_config(
@@ -681,6 +715,14 @@ def add_log_config(self, monitor_name, log_config, force_add=False):
         return log_config
 
     def update_log_configs_on_path(self, path, monitor_name, log_configs):
+        log.log(
+            scalyr_logging.DEBUG_LEVEL_0,
+            (
+                "log_config_for_container update_log_configs_on_path path=%s monitor_name=%s log_configs=%s"
+                % (path, monitor_name, len(log_configs))
+            )
+        )
+
         worker_ids = []
         new_log_configs = []
         for log_config in log_configs:

tests/e2e/scalyr-agent-2-daemonset.yaml

@@ -1,3 +1,47 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: scalyr-config-agent-d
+  namespace: scalyr
+data:
+  k8s_logs.json: |-
+    {
+      "k8s_logs": [
+        {
+          "attributes": {
+            "container_name": "${k8s_container_name}"}
+        }
+      ]
+    }
+  api-key.json: |-
+    {
+       "import_vars": [ "SCALYR_API_KEY" ],
+       "api_key": "$SCALYR_API_KEY"
+      }
+  k8s_events.json: |-
+    {
+      "monitors": [
+        {
+          "module": "scalyr_agent.builtin_monitors.kubernetes_events_monitor"
+        }
+      ]
+    }
+  docker.json: |-
+    {
+      "monitors":[
+        {
+          "module": "scalyr_agent.builtin_monitors.kubernetes_monitor",
+          "stop_agent_on_failure": true
+        }
+      ]
+    }
+  scalyr-server.json: |-
+    {
+        "import_vars": [ { "var": "SCALYR_SERVER", "default": "https://agent.scalyr.com" } ],
+         "scalyr_server": "$SCALYR_SERVER"
+    }
+---
+---
 apiVersion: apps/v1
 kind: DaemonSet
 metadata:
@@ -63,6 +107,8 @@ spec:
         terminationMessagePath: /dev/termination-log
         terminationMessagePolicy: File
         volumeMounts:
+        - name: scalyr-config-agent-d-vol
+          mountPath: /etc/scalyr-agent-2/agent.d
         - mountPath: /var/lib/docker/containers
           name: varlibdockercontainers
           readOnly: true
@@ -101,6 +147,9 @@ spec:
       serviceAccountName: scalyr-service-account
       terminationGracePeriodSeconds: 30
       volumes:
+      - name: scalyr-config-agent-d-vol
+        configMap:
+          name: scalyr-config-agent-d
       - hostPath:
           path: /var/lib/docker/containers
           type: ""