[roshanprabu]

Summary

Fixes one of three bugs reported in #68240 (the one I could verify is still present and fix with high confidence -- see scope note below).

StackdriverRemoteLogIO.processors' proc closure calls _transport.send(...) with no error handling:

_transport.send(record, str(msg.get("event", "")), resource=self.resource, labels=labels)
return event

This processor is installed for every supervised component (scheduler, dag-processor, triggerer, workers) whenever REMOTE_TASK_LOG routes through Stackdriver. Any IAM permission error, gRPC connectivity failure, or quota exception from Cloud Logging propagates straight out of the structlog processor and takes down the whole process -- reported as a dag-processor pod stuck in CrashLoopBackOff on every log emit when its Kubernetes Service Account lacked the logging.logEntries.create IAM binding.

The read path already has exactly this guard, a few lines down in the same file, with a comment explaining the rationale:

try:
    messages, end_of_log, next_page_token = self.io.read_logs(log_filter, next_page_token, all_pages)
except Exception:
    # Cloud Logging unavailable / IAM glitch / gRPC error. Without a guard, the
    # exception used to propagate up as HTTP 500 from the log viewer. ...
    _logger.exception(...)

The write path was simply missing the equivalent. This PR adds it, reusing the same _logger (already defined in this file specifically for handler-internal failures) and logging a warning instead of raising, since log delivery here is best-effort.

Scope note: the linked issue reports three bugs. I traced the other two (empty labels when record.task_instance is unset in supervisor context, and read() filtering on logical_date) against current main and found the code has moved on since the report -- there's now a fallback in proc() that reads labels from the event dict when no task_instance is present (see test_processors_fallback_to_event_labels), and read()'s ti: RuntimeTI already exposes logical_date locally without a DB round-trip. I wasn't confident those two are still bugs as described, so this PR only covers the third (transport.send() unguarded), which I could verify precisely.

Test plan

• Added test_processors_survives_transport_send_failure: mocks transport.send() to raise, asserts the processor doesn't propagate the exception.
• Verified the test actually catches the bug: reverted the source fix and confirmed the new test fails with Exception: IAM permission denied propagating out of proc(); re-applied the fix and confirmed it passes.
• Ran the full file: pytest tests/unit/google/cloud/log/test_stackdriver_task_handler.py -- 33 passed.
• ruff check and ruff format --check pass on both changed files.

[olayiwola-gbenga]

Hi @roshanprabu Please look into failing check

[shahar1]

Thanks for your contribution! Two requests:

1. Resolve conflicts and make the appropriate adjustments as classes have been renamed.
2. Add AI attribution checkboxes to the PR's body.

[shahar1]

Thanks for working on this. With the default BackgroundThreadTransport, which is what the stackdriver:// wiring, from_config and the deprecated alias all use, send() only puts the entry on an in-memory queue. The Cloud Logging API call happens in the library's background worker, which already catches and logs commit errors (_safely_commit_batch), so the new except branch can't run in supported configurations. It would only fire if a custom Python logging config passed SyncTransport explicitly; remote_task_handler_kwargs can't select one. The reporter's stack trace was never provided, and an earlier PR for the same fix (#68250) was closed for the same reason, so I'm closing this one. If you can reproduce the crash with the default transport, please open an issue with the traceback and we'll look at the real failure path.

---

Drafted-by: Claude Code (Opus 5.5); reviewed by @shahar1 before posting