Merge pull request PrefectHQ#3179 from PrefectHQ/logs-updates

joshmeek · web-flow · commit 389eabd68238 · 2020-08-18T15:06:08.000-04:00
Logging updates
diff --git a/changes/pr3179.yaml b/changes/pr3179.yaml
@@ -0,0 +1,5 @@
+enhancement:
+  - "Cloud logger now responds to logging level - [#3179](https://github.com/PrefectHQ/prefect/pull/3179)"
+
+breaking:
+  - "Logging level in Cloud now defaults to INFO - [#3179](https://github.com/PrefectHQ/prefect/pull/3179)"
diff --git a/setup.cfg b/setup.cfg
@@ -4,6 +4,7 @@ norecursedirs = *.egg-info .git .mypy_cache node_modules .pytest_cache .vscode
 env =
     SKIP_DOCKER_ENVIRONMENT_TESTS = True
     PREFECT__USER_CONFIG_PATH=""
+    PREFECT__BACKEND="cloud"
 usedevelop = True
 
 [isort]
diff --git a/src/prefect/agent/docker/agent.py b/src/prefect/agent/docker/agent.py
@@ -450,7 +450,7 @@ def populate_env_vars(self, flow_run: GraphQLResult) -> dict:
             "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,  # type: ignore
             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
             "PREFECT__LOGGING__LOG_TO_CLOUD": str(self.log_to_cloud).lower(),
-            "PREFECT__LOGGING__LEVEL": "DEBUG",
+            "PREFECT__LOGGING__LEVEL": config.logging.level,
             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
             **self.env_vars,
diff --git a/src/prefect/agent/fargate/agent.py b/src/prefect/agent/fargate/agent.py
@@ -604,7 +604,7 @@ def _create_task_definition(
                         "name": "PREFECT__LOGGING__LOG_TO_CLOUD",
                         "value": str(self.log_to_cloud).lower(),
                     },
-                    {"name": "PREFECT__LOGGING__LEVEL", "value": "DEBUG"},
+                    {"name": "PREFECT__LOGGING__LEVEL", "value": config.logging.level},
                     {
                         "name": "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS",
                         "value": "prefect.engine.cloud.CloudFlowRunner",
diff --git a/src/prefect/agent/kubernetes/agent.py b/src/prefect/agent/kubernetes/agent.py
@@ -191,6 +191,7 @@ def replace_job_spec_yaml(self, flow_run: GraphQLResult, image: str) -> dict:
         env[4]["value"] = os.getenv("NAMESPACE", "default")
         env[5]["value"] = str(self.labels)
         env[6]["value"] = str(self.log_to_cloud).lower()
+        env[7]["value"] = config.logging.level
 
         # append all user provided values
         for key, value in self.env_vars.items():
diff --git a/src/prefect/agent/kubernetes/job_spec.yaml b/src/prefect/agent/kubernetes/job_spec.yaml
@@ -29,10 +29,10 @@ spec:
               value: PREFECT__CLOUD__AGENT__LABELS
             - name: PREFECT__LOGGING__LOG_TO_CLOUD
               value: PREFECT__LOGGING__LOG_TO_CLOUD
+            - name: PREFECT__LOGGING__LEVEL
+              value: "INFO"
             - name: PREFECT__CLOUD__USE_LOCAL_SECRETS
               value: "false"
-            - name: PREFECT__LOGGING__LEVEL
-              value: "DEBUG"
             - name: PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS
               value: "prefect.engine.cloud.CloudFlowRunner"
             - name: PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS
diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py
@@ -191,7 +191,7 @@ def populate_env_vars(self, flow_run: GraphQLResult) -> dict:
             "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,  # type: ignore
             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
             "PREFECT__LOGGING__LOG_TO_CLOUD": str(self.log_to_cloud).lower(),
-            "PREFECT__LOGGING__LEVEL": "DEBUG",
+            "PREFECT__LOGGING__LEVEL": config.logging.level,
             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
             **self.env_vars,
diff --git a/src/prefect/engine/cloud/flow_runner.py b/src/prefect/engine/cloud/flow_runner.py
@@ -112,7 +112,7 @@ def call_runner_target_handlers(self, old_state: State, new_state: State) -> Sta
             )
         except Exception as exc:
             msg = "Exception raised while calling state handlers: {}".format(repr(exc))
-            self.logger.debug(msg)
+            self.logger.exception(msg)
             if raise_on_exception:
                 raise exc
             new_state = Failed(msg, result=exc)
@@ -143,7 +143,7 @@ def call_runner_target_handlers(self, old_state: State, new_state: State) -> Sta
             )
             new_state = state
         except Exception as exc:
-            self.logger.debug(
+            self.logger.exception(
                 "Failed to set flow state with error: {}".format(repr(exc))
             )
             raise ENDRUN(state=new_state)
@@ -290,7 +290,7 @@ def run(
             time_remaining = max(
                 (end_state.start_time - pendulum.now("utc")).total_seconds(), 0
             )
-            self.logger.info(
+            self.logger.debug(
                 (
                     f"Flow run is in a Queued state. Sleeping for at most {time_remaining:.2f} "
                     f"seconds and attempting to run again."
diff --git a/src/prefect/engine/flow_runner.py b/src/prefect/engine/flow_runner.py
@@ -347,7 +347,6 @@ def set_flow_to_running(self, state: State) -> State:
             - ENDRUN: if the flow is not pending or running
         """
         if state.is_pending():
-            self.logger.info("Starting flow run.")
             return Running(message="Running flow.")
         elif state.is_running():
             return state
diff --git a/src/prefect/environments/execution/dask/job.yaml b/src/prefect/environments/execution/dask/job.yaml
@@ -41,7 +41,7 @@ spec:
             - name: PREFECT__LOGGING__LOG_TO_CLOUD
               value: "true"
             - name: PREFECT__LOGGING__LEVEL
-              value: "DEBUG"
+              value: "INFO"
             - name: PREFECT__DEBUG
               value: "true"
             - name: DASK_DISTRIBUTED__SCHEDULER__WORK_STEALING
diff --git a/src/prefect/environments/execution/dask/k8s.py b/src/prefect/environments/execution/dask/k8s.py
@@ -486,6 +486,10 @@ def _populate_scheduler_spec_yaml(self, yaml_obj: dict, docker_name: str) -> dic
                 "name": "PREFECT__LOGGING__LOG_TO_CLOUD",
                 "value": str(prefect.config.logging.log_to_cloud).lower(),
             },
+            {
+                "name": "PREFECT__LOGGING__LEVEL",
+                "value": str(prefect.config.logging.level),
+            },
             {
                 "name": "PREFECT__LOGGING__EXTRA_LOGGERS",
                 "value": self._extra_loggers(),
@@ -546,6 +550,10 @@ def _populate_worker_spec_yaml(self, yaml_obj: dict) -> dict:
                 "name": "PREFECT__LOGGING__LOG_TO_CLOUD",
                 "value": str(prefect.config.logging.log_to_cloud).lower(),
             },
+            {
+                "name": "PREFECT__LOGGING__LEVEL",
+                "value": str(prefect.config.logging.level),
+            },
             {
                 "name": "PREFECT__LOGGING__EXTRA_LOGGERS",
                 "value": self._extra_loggers(),
diff --git a/src/prefect/utilities/logging.py b/src/prefect/utilities/logging.py
@@ -129,6 +129,15 @@ def emit(self, record) -> None:  # type: ignore
             assert isinstance(self.client, Client)  # mypy assert
 
             record_dict = record.__dict__.copy()
+
+            # ensures emitted logs respect configured logging level
+            config_level = getattr(
+                logging, prefect.context.config.logging.level, logging.INFO
+            )
+
+            if record_dict["levelno"] < config_level:
+                return
+
             # remove potentially non-json serializable formatting args
             record_dict.pop("args", None)
 
@@ -213,6 +222,9 @@ def _create_logger(name: str) -> logging.Logger:
     logger.addHandler(handler)
     logger.setLevel(context.config.logging.level)
 
+    # we set the cloud handler to DEBUG level
+    # but the handler itself will dynamically respond
+    # to the configured level in the emit() method call
     cloud_handler = CloudHandler()
     cloud_handler.setLevel("DEBUG")
     logger.addHandler(cloud_handler)
diff --git a/tests/agent/test_docker_agent.py b/tests/agent/test_docker_agent.py
@@ -176,7 +176,7 @@ def test_populate_env_vars(monkeypatch, cloud_api):
             "PREFECT__CONTEXT__FLOW_ID": "foo",
             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
             "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
-            "PREFECT__LOGGING__LEVEL": "DEBUG",
+            "PREFECT__LOGGING__LEVEL": "INFO",
             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
         }
@@ -212,7 +212,7 @@ def test_populate_env_vars_includes_agent_labels(monkeypatch, cloud_api):
             "PREFECT__CONTEXT__FLOW_ID": "foo",
             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
             "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
-            "PREFECT__LOGGING__LEVEL": "DEBUG",
+            "PREFECT__LOGGING__LEVEL": "INFO",
             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
         }
diff --git a/tests/agent/test_fargate_agent.py b/tests/agent/test_fargate_agent.py
@@ -1026,7 +1026,7 @@ def test_deploy_flow_register_task_definition_all_args(
                 {"name": "PREFECT__CLOUD__AGENT__LABELS", "value": "[]"},
                 {"name": "PREFECT__CLOUD__USE_LOCAL_SECRETS", "value": "false"},
                 {"name": "PREFECT__LOGGING__LOG_TO_CLOUD", "value": "true"},
-                {"name": "PREFECT__LOGGING__LEVEL", "value": "DEBUG"},
+                {"name": "PREFECT__LOGGING__LEVEL", "value": "INFO"},
                 {
                     "name": "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS",
                     "value": "prefect.engine.cloud.CloudFlowRunner",
@@ -1167,7 +1167,7 @@ def test_deploy_flows_includes_agent_labels_in_environment(
                     "name": "PREFECT__LOGGING__LOG_TO_CLOUD",
                     "value": str(not flag).lower(),
                 },
-                {"name": "PREFECT__LOGGING__LEVEL", "value": "DEBUG"},
+                {"name": "PREFECT__LOGGING__LEVEL", "value": "INFO"},
                 {
                     "name": "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS",
                     "value": "prefect.engine.cloud.CloudFlowRunner",
@@ -1265,7 +1265,7 @@ def test_deploy_flows_enable_task_revisions_no_tags(monkeypatch, cloud_api):
                     {"name": "PREFECT__CLOUD__AGENT__LABELS", "value": "[]"},
                     {"name": "PREFECT__CLOUD__USE_LOCAL_SECRETS", "value": "false"},
                     {"name": "PREFECT__LOGGING__LOG_TO_CLOUD", "value": "true"},
-                    {"name": "PREFECT__LOGGING__LEVEL", "value": "DEBUG"},
+                    {"name": "PREFECT__LOGGING__LEVEL", "value": "INFO"},
                     {
                         "name": "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS",
                         "value": "prefect.engine.cloud.CloudFlowRunner",
@@ -1646,7 +1646,7 @@ def test_deploy_flows_enable_task_revisions_with_external_kwargs(
                     },
                     {"name": "PREFECT__CLOUD__USE_LOCAL_SECRETS", "value": "false"},
                     {"name": "PREFECT__LOGGING__LOG_TO_CLOUD", "value": "false"},
-                    {"name": "PREFECT__LOGGING__LEVEL", "value": "DEBUG"},
+                    {"name": "PREFECT__LOGGING__LEVEL", "value": "INFO"},
                     {
                         "name": "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS",
                         "value": "prefect.engine.cloud.CloudFlowRunner",
diff --git a/tests/agent/test_local_agent.py b/tests/agent/test_local_agent.py
@@ -164,7 +164,7 @@ def test_populate_env_vars(cloud_api):
             "PREFECT__CONTEXT__FLOW_ID": "foo",
             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
             "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
-            "PREFECT__LOGGING__LEVEL": "DEBUG",
+            "PREFECT__LOGGING__LEVEL": "INFO",
             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
         }
@@ -205,7 +205,7 @@ def test_populate_env_vars_includes_agent_labels(cloud_api):
             "PREFECT__CONTEXT__FLOW_ID": "foo",
             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
             "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
-            "PREFECT__LOGGING__LEVEL": "DEBUG",
+            "PREFECT__LOGGING__LEVEL": "INFO",
             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
         }
diff --git a/tests/engine/cloud/test_cloud_flow_runner.py b/tests/engine/cloud/test_cloud_flow_runner.py
@@ -540,7 +540,7 @@ def log_stuff():
 
     time.sleep(0.75)
     logs = [log for call in calls for log in call[0]]
-    assert len(logs) >= 6  # actual number of logs
+    assert len(logs) >= 5  # actual number of logs
 
     loggers = [c["name"] for c in logs]
     assert set(loggers) == {
diff --git a/tests/engine/test_flow_runner.py b/tests/engine/test_flow_runner.py
@@ -1375,7 +1375,7 @@ def log_stuff():
 
     time.sleep(0.75)
     assert len(calls) >= 1
-    assert len([log for call in calls for log in call[0]]) == 6  # actual number of logs
+    assert len([log for call in calls for log in call[0]]) == 5  # actual number of logs
 
     loggers = [log["name"] for call in calls for log in call[0]]
     assert set(loggers) == {
diff --git a/tests/environments/execution/test_dask_k8s_environment.py b/tests/environments/execution/test_dask_k8s_environment.py
@@ -415,8 +415,9 @@ def test_populate_custom_worker_spec_yaml(log_flag):
     assert env[5]["value"] == "prefect.engine.cloud.CloudTaskRunner"
     assert env[6]["value"] == "prefect.engine.executors.DaskExecutor"
     assert env[7]["value"] == str(log_flag).lower()
+    assert env[8]["value"] == "INFO"
     assert (
-        env[8]["value"]
+        env[9]["value"]
         == "['test_logger', 'dask_kubernetes.core', 'distributed.deploy.adaptive']"
     )
 
@@ -462,8 +463,9 @@ def test_populate_custom_scheduler_spec_yaml(log_flag):
     assert env[7]["value"] == "prefect.engine.cloud.CloudTaskRunner"
     assert env[8]["value"] == "prefect.engine.executors.DaskExecutor"
     assert env[9]["value"] == str(log_flag).lower()
+    assert env[10]["value"] == "INFO"
     assert (
-        env[10]["value"]
+        env[11]["value"]
         == "['test_logger', 'dask_kubernetes.core', 'distributed.deploy.adaptive']"
     )
 
diff --git a/tests/utilities/test_logging.py b/tests/utilities/test_logging.py
@@ -161,6 +161,33 @@ def test_cloud_handler_formats_messages_and_removes_args(caplog, monkeypatch):
         logger.handlers = []
 
 
+def test_cloud_handler_emit_responds_to_config(caplog, monkeypatch):
+    monkeypatch.setattr("prefect.client.Client", MagicMock)
+    client = MagicMock()
+    try:
+        with utilities.configuration.set_temporary_config(
+            {"logging.log_to_cloud": True, "logging.level": "DEBUG"}
+        ):
+            logger = utilities.logging.configure_logging(testing=True)
+            assert hasattr(logger.handlers[-1], "client")
+            logger.handlers[-1].client = client
+
+            child_logger = logger.getChild("sub-test")
+
+            with utilities.configuration.set_temporary_config(
+                {"logging.log_to_cloud": True, "logging.level": "INFO"}
+            ):
+                child_logger.debug("heres a log")
+
+            time.sleep(0.75)
+
+            assert not client.write_run_logs.called
+    finally:
+        # reset root_logger
+        logger = utilities.logging.configure_logging(testing=True)
+        logger.handlers = []
+
+
 def test_remote_handler_ships_json_payloads(caplog, monkeypatch):
     monkeypatch.setattr("prefect.client.Client", MagicMock)
     client = MagicMock()

Original file line number	Diff line number	Diff line change
`@@ -604,7 +604,7 @@ def _create_task_definition(`
`604`	`604`	`"name": "PREFECT__LOGGING__LOG_TO_CLOUD",`
`605`	`605`	`"value": str(self.log_to_cloud).lower(),`
`606`	`606`	`},`
`607`		`- {"name": "PREFECT__LOGGING__LEVEL", "value": "DEBUG"},`
	`607`	`+ {"name": "PREFECT__LOGGING__LEVEL", "value": config.logging.level},`
`608`	`608`	`{`
`609`	`609`	`"name": "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS",`
`610`	`610`	`"value": "prefect.engine.cloud.CloudFlowRunner",`
Original file line number	Diff line number	Diff line change
`@@ -112,7 +112,7 @@ def call_runner_target_handlers(self, old_state: State, new_state: State) -> Sta`
`112`	`112`	`)`
`113`	`113`	`except Exception as exc:`
`114`	`114`	`msg = "Exception raised while calling state handlers: {}".format(repr(exc))`
`115`		`- self.logger.debug(msg)`
	`115`	`+ self.logger.exception(msg)`
`116`	`116`	`if raise_on_exception:`
`117`	`117`	`raise exc`
`118`	`118`	`new_state = Failed(msg, result=exc)`
`@@ -143,7 +143,7 @@ def call_runner_target_handlers(self, old_state: State, new_state: State) -> Sta`
`143`	`143`	`)`
`144`	`144`	`new_state = state`
`145`	`145`	`except Exception as exc:`
`146`		`- self.logger.debug(`
	`146`	`+ self.logger.exception(`
`147`	`147`	`"Failed to set flow state with error: {}".format(repr(exc))`
`148`	`148`	`)`
`149`	`149`	`raise ENDRUN(state=new_state)`
`@@ -290,7 +290,7 @@ def run(`
`290`	`290`	`time_remaining = max(`
`291`	`291`	`(end_state.start_time - pendulum.now("utc")).total_seconds(), 0`
`292`	`292`	`)`
`293`		`- self.logger.info(`
	`293`	`+ self.logger.debug(`
`294`	`294`	`(`
`295`	`295`	`f"Flow run is in a Queued state. Sleeping for at most {time_remaining:.2f} "`
`296`	`296`	`f"seconds and attempting to run again."`