feat: add custom ETL connector support (YET-1178)

[sassmith]

Summary

• Add runtime-loaded custom ETL connector infrastructure at apollo/integrations/custom_etl/, mirroring the existing custom (warehouse) connector scaffolding
• CustomEtlProxyClient wraps Connector(BaseEtlConnector) modules loaded from /opt/custom-etl-connectors/{name}/, exposing fetch_etl_assets and fetch_etl_runs methods that match the agent calls in data-collector PR #2390
• Factory routes custom-etl-connector-<hash> connection types through the new proxy client (gated by MCD_CUSTOM_CONNECTORS_ENABLED)
• Discovery endpoints (/api/v1/agent/connectors/types and /api/v1/agent/custom-connectors/manifests) now return custom ETL connector types and manifests alongside existing native/custom entries

Related issues and PRs

• YET-1178
• data-collector PR #2390 (DC-side agent calls)

Test plan

• 40 new tests in `tests/test_custom_etl_proxy_client.py` covering loader, serialization, proxy client, discovery, agent integration, and factory routing
• Existing `test_custom_proxy_client.py` (33 tests) updated and passing
• `test_proxy_client_factory.py` passing
• pyright clean on new modules (0 errors)
• CI passes
• Ran `/code-review`

🤖 Generated with Claude Code

[linear]

YET-1178

[genehynson]

Cross-repo contract check (vs saas-serverless #4279 normalizer + pycarlo wire envelope) — possible asset-drop bug.

The ETL metadata wire envelope is asymmetric in pycarlo (pycarlo/features/ingestion/etl/models.py):

• metadata events are wrapped: build_etl_metadata_payload emits [EtlMetadataEvent(etl_asset=e).to_dict() for e in events] → each event is {"etl_asset": {job_source_id, name, ...}}. The EtlMetadataEvent.etl_asset field is required: True on the wire schema.
• run events are flat: build_etl_runs_payload emits [e.to_dict() for e in events] → {job_source_id, run_source_id, ...} with no wrapper.

The saas-serverless normalizer matches this exactly — asset path does wire_event.get("etl_asset") and skips the event if it's missing; runs path reads fields top-level.

This PR's CustomEtlProxyClient._serialize reflects the EtlAsset object flat for both methods, so fetch_etl_assets returns {"job_source_id": ..., "name": ...} without the etl_asset wrapper. End result: every asset event silently drops at the normalizer (.get("etl_asset") → None → skip). Runs are unaffected (flat is correct for them).

Suggested fix: serialize assets through the pycarlo envelope rather than the generic _serialize — e.g. EtlMetadataEvent(etl_asset=e).to_dict() (or build_etl_metadata_payload) for fetch_etl_assets, keeping runs flat. Since the customer connector's fetch_metadata already returns List[EtlAsset] (pycarlo type), round-tripping through pycarlo's own serializers also avoids field-name/null-stripping drift from the reflection-based _serialize.

Worth a unit test asserting fetch_etl_assets output has the etl_asset key, to lock the contract against the normalizer.

[sassmith]

Code Review — 8 findings (7 fixed, 1 skipped)

Reviewed: 2026-06-05 UTC | Reviewers: security, correctness, testing, devdocs | Scope: full PR

#	Severity	Finding	Status
F1	ISSUE	Credential validation ordering — module loaded and Connector() instantiated before credentials checked	✅ Moved credential validation before module load
F2	ISSUE	_serialize strips None from dataclass/object dicts but not from plain dicts	✅ Added None-stripping to dict branch for consistency
F3	ISSUE	Module not registered in sys.modules — re-executes top-level code on each load	✅ Added module-level cache to avoid re-execution
F4	SUGGESTION	Missing test for setup_connection failure path	✅ Added test for setup_connection failure
F5	SUGGESTION	Missing test for close() exception suppression	✅ Added test for close() exception suppression
F6	SUGGESTION	Registry cache state bleed between tests	✅ Added tearDown to reset registry cache
F7	SUGGESTION	Path traversal risk in load_connector_module	⏭️ Skipped — matches existing warehouse connector pattern
F8	NIT	Stale _fetch_run_details reference in fetch_etl_runs docstring	✅ Fixed docstring reference

👉 Fix commit