[Logger] Enhance formatter creation (#5561)

mlrun/config.py

@@ -1406,14 +1406,14 @@ def read_env(env=None, prefix=env_prefix):
     if log_formatter_name := config.get("log_formatter"):
         import mlrun.utils.logger
 
-        log_formatter = mlrun.utils.create_formatter_instance(
+        log_formatter = mlrun.utils.resolve_formatter_by_kind(
             mlrun.utils.FormatterKinds(log_formatter_name)
         )
         current_handler = mlrun.utils.logger.get_handler("default")
         current_formatter_name = current_handler.formatter.__class__.__name__
-        desired_formatter_name = log_formatter.__class__.__name__
+        desired_formatter_name = log_formatter.__name__
         if current_formatter_name != desired_formatter_name:
-            current_handler.setFormatter(log_formatter)
+            current_handler.setFormatter(log_formatter())
 
     # The default function pod resource values are of type str; however, when reading from environment variable numbers,
     # it converts them to type int if contains only number, so we want to convert them to str.

mlrun/utils/logger.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import logging
+import typing
 from enum import Enum
 from sys import stdout
 from traceback import format_exception
@@ -221,11 +222,15 @@ class FormatterKinds(Enum):
     JSON = "json"
 
 
-def create_formatter_instance(formatter_kind: FormatterKinds) -> logging.Formatter:
+def resolve_formatter_by_kind(
+    formatter_kind: FormatterKinds,
+) -> type[
+    typing.Union[HumanReadableFormatter, HumanReadableExtendedFormatter, JSONFormatter]
+]:
     return {
-        FormatterKinds.HUMAN: HumanReadableFormatter(),
-        FormatterKinds.HUMAN_EXTENDED: HumanReadableExtendedFormatter(),
-        FormatterKinds.JSON: JSONFormatter(),
+        FormatterKinds.HUMAN: HumanReadableFormatter,
+        FormatterKinds.HUMAN_EXTENDED: HumanReadableExtendedFormatter,
+        FormatterKinds.JSON: JSONFormatter,
     }[formatter_kind]
 
 
@@ -243,11 +248,11 @@ def create_logger(
     logger_instance = Logger(level, name=name, propagate=False)
 
     # resolve formatter
-    formatter_instance = create_formatter_instance(
+    formatter_instance = resolve_formatter_by_kind(
         FormatterKinds(formatter_kind.lower())
     )
 
     # set handler
-    logger_instance.set_handler("default", stream or stdout, formatter_instance)
+    logger_instance.set_handler("default", stream or stdout, formatter_instance())
 
     return logger_instance

server/api/runtime_handlers/daskjob.py

@@ -350,6 +350,9 @@ def get_env_name(env_: Union[client.V1EnvVar, dict]) -> str:
     if spec.extra_pip:
         env.append(spec.extra_pip)
 
+    # remove duplicates by name
+    env = list({get_env_name(spec_env): spec_env for spec_env in env}.values())
+
     pod_labels = get_resource_labels(function, scrape_metrics=config.scrape_metrics)
 
     worker_args = ["dask", "worker"]

tests/api/runtimes/test_dask.py

@@ -449,6 +449,8 @@ def test_enrich_dask_cluster(self):
                 env=[
                     {"name": "MLRUN_NAMESPACE", "value": "other-namespace"},
                     k8s_client.V1EnvVar(name="MLRUN_TAG", value="latest"),
+                    {"name": "TEST_DUP", "value": "A"},
+                    {"name": "TEST_DUP", "value": "B"},
                 ],
             ),
         )
@@ -472,6 +474,7 @@ def test_enrich_dask_cluster(self):
             {"name": "MLRUN_DEFAULT_PROJECT", "value": "project"},
             {"name": "MLRUN_NAMESPACE", "value": "test-namespace"},
             k8s_client.V1EnvVar(name="MLRUN_TAG", value="latest"),
+            {"name": "TEST_DUP", "value": "A"},
         ]
         expected_labels = {
             "mlrun/project": "project",