howardyoo commented on code in PR #40802:
URL: https://github.com/apache/airflow/pull/40802#discussion_r1683580208
##########
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()")
Review Comment:
> Non-blocking question, I don't have the end-user experience to make this
call so fee free to disagree and ignore: I get you are using the parenthesis at
the end of the name to indicate the event was a method that was called. It
makes sense. Is that some kind of standard? Would there be any down side to
dropping the parenthesis here and elsewhere? It seems like that would result in
a cleaner UX when the user is viewing these.
No, this is not a standard, and we can definitely drop the parenthesis. I'll
drop them, since less is always good!
--
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]
