Encode Celery tasks and results as JSON - #850
Conversation
json.dumps writes bare NaN and Infinity tokens by default. SQLite stores them without complaint but PostgreSQL validates on insert and rejects the whole row, so an execution could succeed while ingesting 0 metric values. Sanitising on the column type covers every JSON write rather than the one reported path. Closes #839
Pickle let anything that could write to the broker run arbitrary code in a worker. The wire format is now JSON and pickle is no longer in accept_content. Only the diagnostic's slug crosses the wire, and the receiver resolves it against its own provider registry. This also fixes a latent bug where the client's conda prefix and HOME travelled inside the pickled diagnostic and overrode the worker's configuration. Closes #847
Message bodies are dominated by the datasets frame in the execution definition and shrink by roughly 80%, which cuts broker memory on a full solve. Override with CELERY_TASK_COMPRESSION and CELERY_RESULT_COMPRESSION, or set either to an empty string to disable. Closes #845
The attrs classes now unstructure and structure through a cattrs converter, so a new field on ExecutionResult or DatasetCollection serialises without hand copying. The hand-rolled codecs remain only where cattrs has no answer: cftime dates, non-finite floats and the dtype-faithful dataset frame. Numeric columns are emitted directly rather than dispatched cell by cell. Also collapses the duplicate lookup loop in provider_by_slug and reads the compression settings through env like their neighbours.
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughChangesCelery transport migration
Core wire serialisation and provider resolution
Non-finite JSON database handling
Deployment failure diagnostics
Sequence Diagram(s)sequenceDiagram
participant CeleryApp
participant KombuRegistry
participant CeleryProducer
participant Broker
participant CeleryWorker
CeleryApp->>KombuRegistry: register ref-json
CeleryProducer->>KombuRegistry: encode and gzip task
CeleryProducer->>Broker: publish task message
Broker->>CeleryWorker: deliver task message
CeleryWorker->>KombuRegistry: decode ref-json
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
- Breaks the remaining hard-wrapped comments and docstrings at clause boundaries. - Notes in the helm README that workers upgrade before clients. - Adds a roundtrip test asserting every ExecutionDefinition field survives the wire, since the structure hook lists the fields by hand.
CondaDiagnosticProvider kept a snapshot of os.environ from construction and passed it to every command it launched. The provider now stores only its own overrides and merges them over the live process environment at launch, so settings applied to a worker reach every execution. HOME still falls back to the conda prefix when the environment does not set it.
* origin/main: (35 commits) test(datasets): isolate parallel worker setting chore: clean up docs feat(datasets): configure parallel fetch workers test(datasets): verify netcdf access stays on caller thread fix(datasets): prepare cache directories before parallel fetch docs: add parallel registry fetch changelog perf(datasets): fetch registry files concurrently chore: rename the exported variable fix: point the missing-data warning at the fetch, not at setup fix(pmp): do not download micromamba while building the provider registry test: re-mint the sea ice baselines for the reference requirement fix: name the reference source type the way a request is read test: harden the offline fixtures after review fix: freeze the shared reference facets fix: address review of the sea ice reference requirement test: prove the offline and read-only paths with blocked sockets docs: add changelog fragment for #846 feat(esmvaltool): select the OSI-450 observations for sea ice area feat(esmvaltool): ingest reference data during provider setup fix(core): address the CodeRabbit review ... # Conflicts: # packages/climate-ref-core/src/climate_ref_core/providers.py # packages/climate-ref-pmp/src/climate_ref_pmp/__init__.py # packages/climate-ref-pmp/tests/unit/test_provider.py
|
Smoke tested this on a Kubernetes deployment (k3s, CNPG Postgres, Dragonfly broker, real CMIP6 data), running the Chart 0.5.0 sets CELERY_ACCEPT_CONTENT: |
["json", "pickle"]The old code ignored it, because The PR removes that default from |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (5)
.github/workflows/packaging.yaml (1)
179-180: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid an unbounded
KEYS('*')scan in the failure path.
KEYS('*')blocks the broker while scanning every key, andr.type(k)adds one request per key. A backlog during a failed deployment can therefore delay or exhaust the diagnostic job before logs are collected. PreferSCANwith bounded batches and a pipeline for the type/length queries.packages/climate-ref-core/src/climate_ref_core/serialisation.py (2)
148-181: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low valueFrame encoding assumes unique column names and
str(dtype)-reconstructible dtypes.Two narrow gaps worth noting:
_decode_frameassigns columns by name, so a frame with duplicate column labels collapses to a single column.str(dtype)is not a faithful round-trip for parameterised extension dtypes (e.g.categoryloses the categories andorderedflag).Neither is exercised by the current tests. If catalogs can never hold either, a short comment saying so would be enough; otherwise consider validating uniqueness and rejecting dtypes that cannot be reconstructed.
27-28: 🗄️ Data Integrity & Integration | 🔵 TrivialConsider a wire-format version marker.
The rollout depends on workers and clients agreeing on this encoding, and today a mismatch surfaces only as
Unknown tagged valueor aKeyErroron a missing field. A single top-level version integer in the envelope would let the receiver fail with an explicit "message produced by a newer client" error, which is much easier to triage during a staged upgrade.packages/climate-ref-core/src/climate_ref_core/providers.py (1)
382-424: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winLookup can import every installed provider, contrary to the docstring.
The name-match ordering only puts the likely candidate first; when the entry-point name does not equal
provider.slug, the loop keeps importing the remaining providers until one matches. On a worker that means pulling in heavy packages (ESMValTool, PMP) just to resolve one slug, and the docstring's "only the provider that matches is ever loaded" no longer holds.Since the name-is-the-slug convention is already assumed for ordering, consider trying the exact name match first and only falling back to a scan (with a debug log noting the fallback), or soften the docstring.
Also note
@functools.cachenever invalidates: entry points installed or providers reloaded later in the same process are not picked up, which mainly bites in tests that manipulate the entry-point set.packages/climate-ref-core/tests/unit/test_providers.py (1)
726-747: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUncovered branch: a non-matching provider that fails to import.
provider_by_slugdeliberately swallowsInvalidProviderExceptionfor entry points whose name differs from the requested slug (providers.pyLines 415-419), and that skip path is the subtle part of the function. Consider a test that patchesimportlib.metadata.entry_pointswith one broken entry point plus a good one and asserts the good provider is still returned. Noteprovider_by_slugis@functools.cached, so such a test needsprovider_by_slug.cache_clear().
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 2945fa07-8eb2-40b8-b4f4-d88dc3fe5571
📒 Files selected for processing (25)
.github/workflows/packaging.yamlchangelog/850.fix.mdchangelog/850.improvement.2.mdchangelog/850.improvement.mdhelm/README.mdhelm/values.yamlpackages/climate-ref-celery/src/climate_ref_celery/app.pypackages/climate-ref-celery/src/climate_ref_celery/celeryconf/base.pypackages/climate-ref-celery/src/climate_ref_celery/serialisation.pypackages/climate-ref-celery/tests/unit/test_app.pypackages/climate-ref-celery/tests/unit/test_celeryconf.pypackages/climate-ref-celery/tests/unit/test_serialisation.pypackages/climate-ref-core/src/climate_ref_core/cmip6_to_cmip7.pypackages/climate-ref-core/src/climate_ref_core/diagnostics.pypackages/climate-ref-core/src/climate_ref_core/providers.pypackages/climate-ref-core/src/climate_ref_core/serialisation.pypackages/climate-ref-core/tests/unit/test_providers.pypackages/climate-ref-core/tests/unit/test_serialisation.pypackages/climate-ref-pmp/src/climate_ref_pmp/__init__.pypackages/climate-ref-pmp/tests/unit/test_provider.pypackages/climate-ref/src/climate_ref/datasets/utils.pypackages/climate-ref/src/climate_ref/models/base.pypackages/climate-ref/tests/unit/models/test_non_finite_json.pypackages/climate-ref/tests/unit/test_offline.pystubs/cftime/__init__.pyi
`_encode_scalar` returned unknown values unchanged, so a `datetime64` or `bytes` cell reached `json.dumps` and raised a generic `TypeError` far from the column that caused it. It now raises, and `_encode_values` names the offending column or the index. Collapses the duplicate scalar dispatch in `to_wire` onto `_encode_scalar`, which already covers every case it handled plus `pd.NA` and `pd.NaT`. Numpy scalars are unwrapped once at the top rather than re-dispatched, which keeps the function under the return limit.
`SeriesMetricValue.values` is assigned from `TSeries.load_from_json(...)`, so a value can arrive as a numpy float rather than a Python one. `isinstance` misses those, and a non-finite float32 reached the JSON column unchanged.
The step runs under errexit, so a failed download or a component with no pod aborted it before the worker logs were collected.
`CELERY_ACCEPT_CONTENT` is read with `env.list`, so it takes a comma separated list rather than the JSON array the chart used to inject.
The diagnostic is a live object that no longer crosses the wire, so it takes no part in equality. Its slug was excluded too, which left two definitions running different diagnostics over the same datasets comparing equal. The slug is now derived at construction, so both sides of the wire hold it as a field and equality compares which diagnostic a definition runs. A diagnostic that has not been registered with a provider yet still has no slug to derive, so building a definition from one stays permitted.
NaN was encoded as JSON null while the infinities were tagged and restored. Inside a dataset frame the recorded column dtype restored it either way, but a NaN on its own has no dtype to be restored from, so it came back as None. Tagging it keeps the two layers honest about their different jobs. The wire stays faithful to the value it was given, and the database column remains the only place a non-finite float becomes null, which is what #839 asked for.
CELERY_TASK_SERIALIZER and CELERY_RESULT_SERIALIZER were env overridable and undocumented, which let a deployment silently start sending pickle again. That is the thing dropping pickle set out to stop. What a process sends is now fixed. What it accepts stays configurable through CELERY_ACCEPT_CONTENT, because a worker may still meet messages another release encoded.
The pickle drain escape hatch does not work. The diagnostic field was renamed on ExecutionDefinition, so an old pickled message restores into an object with neither slot set and execution_slug raises AttributeError. Adding pickle back to CELERY_ACCEPT_CONTENT gets the body decoded and then the task fails anyway. Making it work would need compat shims that resurrect the pickled provider and the environment snapshot this release removed, so the docs now say what is true. Purge the queues on upgrade and re-solve to requeue the outstanding work. Also documents the Celery wire format settings outside the helm chart, and drops a duplicate CELERY_ACCEPT_CONTENT row from the chart README.
changelog/README.md asks for phrases in the past tense, and 848.improvement.md already followed it while the 850 fragments did not. Also collapses them to a single paragraph, which the same file prefers for anything but a feature.
Sorting on a bool to float the name match to the front needed a comment to be readable. Two explicit lists say the same thing on their own. Covers the fallbacks the comment describes, none of which had a test: the named entry point is the only one imported when it matches, a provider whose name and slug disagree is still found, a broken provider we were not asked for is skipped rather than masking the wanted one, and failing to import the provider actually asked for propagates. provider_by_slug is cached for the life of the process, so the tests clear it or a mocked set of entry points is answered from an earlier test.
The comment survived the rewrite from a loop into a per-file callback. There is no next file to move onto now, the callback returns.
Asserting on registry._encoders couples the test to kombu private state. Encoding and decoding a payload proves the same thing through the public interface.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
helm/README.md (1)
209-221: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider calling out the JSON-array pitfall explicitly.
The comma-separated format is documented correctly, but the smoke test on chart 0.5.0 showed the real-world failure mode: a chart supplying
CELERY_ACCEPT_CONTENTas a JSON array string (e.g.["json", "pickle"]) gets mis-parsed byenv.list()into bogus serializer names, causingSerializerNotInstalled. Since the fix for the active default lives in the chart repo (per the PR comments), a short note here warning users with custom overrides not to pass a JSON-array-formatted value would help avoid re-hitting this exact issue.📝 Suggested addition
Tasks and results are encoded as JSON (`ref-json`). What this release sends is fixed, so `CELERY_ACCEPT_CONTENT` only widens what a worker will decode. + +Set `CELERY_ACCEPT_CONTENT` as a comma-separated string (e.g. `json,ref-json`), not a JSON array +(e.g. `["json", "ref-json"]`). Some Helm chart versions supply the latter, which parses into +invalid serializer names and causes workers to fail with `SerializerNotInstalled`.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 216da35d-d703-4d5c-950d-435c29d05d31
📒 Files selected for processing (15)
changelog/850.fix.mdchangelog/850.improvement.2.mdchangelog/850.improvement.mddocs/configuration.mdhelm/README.mdhelm/values.yamlpackages/climate-ref-celery/src/climate_ref_celery/celeryconf/base.pypackages/climate-ref-celery/tests/unit/test_app.pypackages/climate-ref-celery/tests/unit/test_celeryconf.pypackages/climate-ref-core/src/climate_ref_core/dataset_registry.pypackages/climate-ref-core/src/climate_ref_core/diagnostics.pypackages/climate-ref-core/src/climate_ref_core/providers.pypackages/climate-ref-core/src/climate_ref_core/serialisation.pypackages/climate-ref-core/tests/unit/test_providers.pypackages/climate-ref-core/tests/unit/test_serialisation.py
🚧 Files skipped from review as they are similar to previous changes (9)
- changelog/850.improvement.2.md
- packages/climate-ref-celery/tests/unit/test_app.py
- packages/climate-ref-core/src/climate_ref_core/diagnostics.py
- packages/climate-ref-celery/tests/unit/test_celeryconf.py
- helm/values.yaml
- packages/climate-ref-celery/src/climate_ref_celery/celeryconf/base.py
- packages/climate-ref-core/src/climate_ref_core/serialisation.py
- packages/climate-ref-core/src/climate_ref_core/providers.py
- packages/climate-ref-core/tests/unit/test_providers.py
Deriving the slug at construction read _provider directly, which broke every ILAMBStandard because it never calls Diagnostic.__init__ and so has no such attribute to read. Reading it with a default keeps construction working for a subclass that skips the base init.
Flower runs from the stock mher/flower image with its own celeryconfig, so it cannot import the REF codec and refused every message once tasks stopped being pickled. The deployment test logged sixteen ContentDisallowed errors for application/x-ref-json. A ref-json body is plain JSON, so the chart registers a codec that reads it with the json module. Tagged values arrive as their raw mappings, which is all the task views need. Pickle is no longer in the accepted list there either, matching the rest of the deployment.
py-spy returned Permission Denied on every pod, so the failure capture collected no stacks. Granting SYS_PTRACE to the containers cannot work, because they run as a non-root user and Kubernetes has no way to set ambient capabilities, so the capability never reaches the process. Widening the Yama scope on the node lets a process dump another running under the same user, which is what the kubectl exec needs.
py-spy never produced a stack. It needs to attach to a process it did not start, which the pods cannot allow because they run as a non-root user and Kubernetes has no way to grant the capability that would permit it. Celery already dumps every thread's stack when it receives SIGUSR1, and the dump goes to the log the next step collects. That needs no download, no copy into each pod and no ptrace, so the node sysctl goes too.
The pull request appeared four times in the compiled changelog, twice in each section. The wire format, the compression that rides on it and the knobs that tune it are one story, so they are now one entry. The Flower codec is not a fix from a reader's point of view, because nothing shipped with Flower broken. What is left of it is the upgrade note for anyone overriding `flower.celeryConfig`. The PostgreSQL fix stays on its own, because it is a database bug that has nothing to do with Celery.
1 participant
Description
Replaces pickle with a JSON wire format for Celery messages, compresses them, and fixes non-finite floats breaking metric ingestion on PostgreSQL.
Closes #839, closes #845, closes #847.
ref-json). Pickle let anything that could write to the broker run arbitrary code in a worker, so it is no longer inaccept_content. A rolling deploy with pickled messages still in flight can setCELERY_ACCEPT_CONTENTto includepickleuntil the queues drain.DASK_SCHEDULERset on worker pods now apply to queued work.CELERY_TASK_COMPRESSIONandCELERY_RESULT_COMPRESSION, or set either to an empty string to disable.Checklist
Please confirm that this pull request has done the following:
changelog/Notes
CondaDiagnosticProviderno longer capturesos.environas instance state. It keeps only its own overrides and merges the live process environment at command launch, completing the fix suggested in CondaDiagnosticProvider pickles a snapshot of os.environ into every task message #847.ref-json. The helm README documents this and the pickle drain escape hatch.Summary by CodeRabbit
json,ref-json(pickle no longer on the wire).nullin the database.