Guard transport.send() in proc() so IAM/gRPC errors don't crash supervised components

[goingforstudying-ctrl]

Fixes Bug 3 from #68240.

What

StackdriverRemoteLogIO.processors → proc() calls transport.send() without
any error handling. In Airflow 3's supervisor model, REMOTE_TASK_LOG applies
to ALL supervised components (scheduler, dag-processor, triggerer, workers).
A single IAM misconfiguration or gRPC error would crash the entire process.

Observed: dag-processor pod enters CrashLoopBackOff on every log emit when the
Kubernetes Service Account lacks the logging.logEntries.create IAM binding.

Fix

Wrap _transport.send() in try/except Exception and emit a
logging.warning instead of propagating. Log delivery is best-effort;
a Cloud Logging error should never kill a task-executing process.

Changes

• Guarded _transport.send() call in proc() with try/except
• New test test_processors_survives_transport_send_failure verifies:
  • proc() returns the event (doesn't crash)
  • warning is logged with the failure details

relates to #68240

[henry3260]

Could you clarify how the observed IAM, gRPC error propagates from _transport.send()? StackdriverRemoteLogIO uses BackgroundThreadTransport by default, whosesend()only enqueues the entry. The network request happens in the background worker, where exceptions frombatch.commit()are already caught.

[shahar1]

Could you clarify how the observed IAM, gRPC error propagates from _transport.send()? StackdriverRemoteLogIO uses BackgroundThreadTransport by default, whosesend()only enqueues the entry. The network request happens in the background worker, where exceptions frombatch.commit()are already caught.

@goingforstudying-ctrl please address Henry's concern, thanks!

[goingforstudying-ctrl]

Closing this one — the label extraction fix was already merged via #68292 by ashb. Henry's point about BackgroundThreadTransport already catching network errors is also correct: send() only enqueues so the described IAM crash scenario doesn't apply. Thanks for the reviews.