howardyoo commented on code in PR #40802:
URL: https://github.com/apache/airflow/pull/40802#discussion_r1683586391
##########
airflow/dag_processing/manager.py:
##########
@@ -562,118 +568,144 @@ def _run_parsing_loop(self):
# in sync mode we need to be told to start a "loop"
self.start_new_processes()
while True:
- loop_start_time = time.monotonic()
- ready = multiprocessing.connection.wait(self.waitables.keys(),
timeout=poll_time)
- self.heartbeat()
- if self._direct_scheduler_conn is not None and
self._direct_scheduler_conn in ready:
- agent_signal = self._direct_scheduler_conn.recv()
-
- self.log.debug("Received %s signal from
DagFileProcessorAgent", agent_signal)
- if agent_signal == DagParsingSignal.TERMINATE_MANAGER:
- self.terminate()
- break
- elif agent_signal == DagParsingSignal.END_MANAGER:
- self.end()
- sys.exit(os.EX_OK)
- elif agent_signal == DagParsingSignal.AGENT_RUN_ONCE:
- # continue the loop to parse dags
- pass
- elif isinstance(agent_signal, CallbackRequest):
- self._add_callback_to_queue(agent_signal)
- else:
- raise ValueError(f"Invalid message {type(agent_signal)}")
-
- if not ready and not self._async_mode:
- # In "sync" mode we don't want to parse the DAGs until we
- # are told to (as that would open another connection to the
- # SQLite DB which isn't a good practice
-
- # This shouldn't happen, as in sync mode poll should block for
- # ever. Lets be defensive about that.
- self.log.warning(
- "wait() unexpectedly returned nothing ready after infinite
timeout (%r)!", poll_time
- )
-
- continue
-
- for sentinel in ready:
- if sentinel is not self._direct_scheduler_conn:
- processor = self.waitables.get(sentinel)
- if processor:
- self._collect_results_from_processor(processor)
- self.waitables.pop(sentinel)
- self._processors.pop(processor.file_path)
-
- if self.standalone_dag_processor:
- self._fetch_callbacks(max_callbacks_per_loop)
- self._scan_stale_dags()
- DagWarning.purge_inactive_dag_warnings()
- refreshed_dag_dir = self._refresh_dag_dir()
-
- self._kill_timed_out_processors()
-
- # Generate more file paths to process if we processed all the
files already. Note for this
- # to clear down, we must have cleared all files found from
scanning the dags dir _and_ have
- # cleared all files added as a result of callbacks
- if not self._file_path_queue:
- self.emit_metrics()
- self.prepare_file_path_queue()
-
- # if new files found in dag dir, add them
- elif refreshed_dag_dir:
- self.add_new_file_path_to_queue()
-
- self._refresh_requested_filelocs()
- self.start_new_processes()
-
- # Update number of loop iteration.
- self._num_run += 1
-
- if not self._async_mode:
- self.log.debug("Waiting for processors to finish since we're
using sqlite")
- # Wait until the running DAG processors are finished before
- # sending a DagParsingStat message back. This means the Agent
- # can tell we've got to the end of this iteration when it sees
- # this type of message
- self.wait_until_finished()
-
- # Collect anything else that has finished, but don't kick off any
more processors
- self.collect_results()
+ with Trace.start_span(span_name="dag_parsing_loop",
component="DagFileProcessorManager") as span:
+ loop_start_time = time.monotonic()
+ ready = multiprocessing.connection.wait(self.waitables.keys(),
timeout=poll_time)
+ self.heartbeat()
+ if span.is_recording():
+ span.add_event(name="heartbeat()")
+ if self._direct_scheduler_conn is not None and
self._direct_scheduler_conn in ready:
+ agent_signal = self._direct_scheduler_conn.recv()
+
+ self.log.debug("Received %s signal from
DagFileProcessorAgent", agent_signal)
+ if agent_signal == DagParsingSignal.TERMINATE_MANAGER:
+ self.terminate()
+ if span.is_recording():
+ span.add_event(name="terminate()")
+ break
+ elif agent_signal == DagParsingSignal.END_MANAGER:
+ self.end()
+ if span.is_recording():
+ span.add_event(name="end()")
Review Comment:
> Same caveat, I haven't seen this in use as a user so feel free to disagree
and resolve. On an instinctual level, it feels like we should add_event before
calling the method. In this case, if heartbeat() fails and causes an issue,
that call to heartbeat wouldn't be noted in the span and that would make it
harder to trace back where it actually ran into the issue, no? I see you did it
that way in some places below, but not everywhere, so maybe there is a reason
for the inconsistency.
I believe the proper ordering should be 1) function was called, and 2) we
record the event. I'll update parts where the event was logged first, but that
should be changed. Good point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
