ci: harden workflows, add region fallback & /gcbrun, and remove legacy CI files

[dborowitz]

Summary

Modernizes the repository's CI workflows, integration test resiliency, packaging metadata, and developer documentation:

1. Integration Test Resiliency (tests/integration/conftest.py)

• Multi-region 503 stockout fallback: Adds _create_session_with_region_fallback, which retries shared session creation across GOOGLE_CLOUD_FALLBACK_REGIONS (defaulting to us-central1, us-east1, us-west1 when MANAGED_SPARK_CONNECT_SUBNET is a short name and no custom regional endpoint is set) when Dataproc returns 503 ServiceUnavailable (including ServiceUnavailable wrapped in RuntimeError by ManagedSparkSession.Builder). Updates _region_state["region"] so all downstream fixtures (test_region, session_name, session_controller_client, session_template_controller_client) stay pinned to the region that succeeded.
• Graceful SIGTERM / SIGINT session cleanup: Registers a SIGTERM handler and finally: cleanup hooks calling an idempotent _terminate_active_session() helper (falling back to SessionControllerClient.terminate_session if a session is still CREATING when a workflow run is cancelled).
• Unit tests (tests/unit/test_integration_conftest.py): Covers candidate region selection, wrapped 503 ServiceUnavailable fallback, non-503 error propagation, and idempotent in-flight session termination.

2. GitHub Actions Hardening, /gcbrun Support & Dependabot (.github/)

• Fork PR integration testing via /gcbrun (.github/workflows/integration-tests.yaml):
  • Runs automatically on push to main, workflow_dispatch, and same-repo pull_request events (excluding dependabot[bot]).
  • Supports maintainer-triggered /gcbrun (or /gcbrun <sha>) comments on fork PRs: verifies collaborator admin/maintain/write permissions and TOCTOU commit timestamps before checking out the PR head SHA, and reports pending/final commit statuses back to the PR head SHA.
• Repository variables (vars.*) & secret masking:
  • Reads non-secret vars.GCP_REGION and vars.GCP_SUBNET (with fallback to secrets.*), and masks project_id and client_email extracted from secrets.GCP_SA_KEY via ::add-mask::.
• Workflow security & reliability (tests.yaml, pyink.yaml, lint-readme.yaml, integration-tests.yaml):
  • Adds least-privilege permissions: contents: read, persist-credentials: false on actions/checkout, job/workflow concurrency groups with cancel-in-progress, explicit timeout-minutes, and full 40-char commit SHA action pins (0 findings under zizmor).
• Dependabot (.github/dependabot.yml): Adds weekly grouped github-actions updates.

3. Legacy CI Removal & Packaging Modernization

• Removes obsolete .kokoro/, .trampolinerc, cloudbuild/, and .dockerignore files.
• Adds PEP 517 [build-system] to pyproject.toml, removes deprecated [bdist_wheel] universal = 1 from setup.cfg, and updates requirements-dev.txt, DEVELOPING.md, and .env.example.

[gemini-code-assist]

Code Review

This pull request simplifies the development and testing environment by removing legacy Kokoro and Cloud Build configurations, updating dependencies to use local editable installs, and recommending uv in the documentation. Crucially, it enhances the integration test suite in tests/integration/conftest.py with a region fallback mechanism to handle 503 ServiceUnavailable stockouts and registers a SIGTERM handler for reliable session cleanup, backed by new unit tests. The review feedback suggests making the exception-traversing helper more robust against non-exception types and wrapping the signal registration in try-except blocks to prevent failures in multi-threaded test environments.

[gemini-code-assist]

To make _is_service_unavailable more robust and adhere to defensive programming practices, we should ensure that current is an instance of BaseException before attempting to access its __cause__ or __context__ attributes. If a user or library manually assigns a non-exception object to __cause__ or __context__, accessing these attributes on a non-exception type in subsequent iterations would raise an AttributeError. Checking isinstance(current, BaseException) prevents this potential crash.

Suggested change

	def _is_service_unavailable(exc):
	"""Return True if exc or its chained cause/context is 503 ServiceUnavailable."""
	current = exc
	visited = set()
	while current is not None and id(current) not in visited:
	visited.add(id(current))
	if isinstance(current, core_exceptions.ServiceUnavailable):
	return True
	current = current.__cause__ or current.__context__
	return False
	def _is_service_unavailable(exc):
	"""Return True if exc or its chained cause/context is 503 ServiceUnavailable."""
	current = exc
	visited = set()
	while isinstance(current, BaseException) and id(current) not in visited:
	visited.add(id(current))
	if isinstance(current, core_exceptions.ServiceUnavailable):
	return True
	current = current.__cause__ or current.__context__
	return False

[gemini-code-assist]

In Python, signal.signal can only be called from the main thread of the process. If these integration tests are executed in a multi-threaded test runner or environment where this fixture runs on a background thread, calling signal.signal will raise a ValueError. Additionally, signal.getsignal can occasionally return None (e.g., if the handler was registered outside of Python), which would cause signal.signal to raise a TypeError during restoration. Wrapping these calls in try-except (ValueError, TypeError) blocks ensures the test suite remains robust and does not crash unexpectedly.

    def _handle_sigterm(signum, frame):
        _terminate_active_session()
        if callable(previous_sigterm):
            previous_sigterm(signum, frame)
        raise KeyboardInterrupt("Received SIGTERM during integration tests")

    previous_sigterm = None
    try:
        previous_sigterm = signal.getsignal(signal.SIGTERM)
        signal.signal(signal.SIGTERM, _handle_sigterm)
    except (ValueError, TypeError):
        pass
    try:
        yield os.environ
    finally:
        _terminate_active_session()
        if previous_sigterm is not None:
            try:
                signal.signal(signal.SIGTERM, previous_sigterm)
            except (ValueError, TypeError):
                pass