[Feature] Consider all node status.

[maoxingda]

Description

astronomer-cosmos/cosmos/operators/_watcher/base.py

Line 49 in 3355910

if node_status in ["success", "failed"]:

Use case/motivation

sql query timeout.

Related issues

No response

Are you willing to submit a PR?

• Yes, I am willing to submit a PR!

[tatiana]

Thanks a lot for pointing this, @maoxingda ! Which status did you face issues with Cosmos Watcher mode? Would you by any chance like to contribute this fix?

[tatiana]

I believe we'll need to change more parts of the code to fix this, such as:

astronomer-cosmos/cosmos/operators/_watcher/triggerer.py

Lines 133 to 166 in 3355910

	async def run(self) -> AsyncIterator[TriggerEvent]:
	logger.info("Starting WatcherTrigger for model: %s", self.model_unique_id)
	while True:
	producer_task_state = await self._get_producer_task_status()
	node_status = await self._parse_node_status()
	if node_status == "success":
	logger.info("Model '%s' succeeded", self.model_unique_id)
	yield TriggerEvent({"status": "success"}) # type: ignore[no-untyped-call]
	return
	elif node_status == "failed":
	logger.warning("Model '%s' failed", self.model_unique_id)
	yield TriggerEvent({"status": "failed", "reason": "model_failed"}) # type: ignore[no-untyped-call]
	return
	elif producer_task_state == "failed":
	logger.error(
	"Watcher producer task '%s' failed before delivering results for model '%s'",
	self.producer_task_id,
	self.model_unique_id,
	)
	yield TriggerEvent({"status": "failed", "reason": "producer_failed"}) # type: ignore[no-untyped-call]
	return
	elif producer_task_state == "success" and node_status is None:
	logger.info(
	"The producer task '%s' succeeded. There is no information about the model '%s' execution.",
	self.producer_task_id,
	self.model_unique_id,
	)
	yield TriggerEvent({"status": "success", "reason": "model_not_run"}) # type: ignore[no-untyped-call]
	return
	# Sleep briefly before re-polling
	await asyncio.sleep(self.poke_interval)
	logger.debug("Polling again for model '%s' status...", self.model_unique_id)

We'll aim to fix this in Cosmos 1.13.1.