feat(inference-logging-client): .log reader + multi-sink outputs (Spark / pandas / CSV / JSONL / text / analyze)#383
Conversation
Adds first-class support for reading asyncloguploader .log containers directly (local disk, file-like, or gs://) into any of six output sinks — Spark, pandas, CSV, JSONL, human-readable text, or a structural analyze report — all backed by one shared decode path so row semantics are identical across sinks. - log_reader.py: new module with iter_log_records / iter_decoded_log_rows and six sink helpers (decode_log_file, decode_log_file_to_pandas, decode_log_file_to_csv/jsonl, write_parsed_log, analyze_log_file). - __init__.py: export the new public API. - cli.py: auto-route .log / gs:// inputs; add --output-format and --strict flags. - pyproject.toml: add [gcs] optional extra pulling google-cloud-storage. - readme.md: consolidated .log documentation into one Quick Start subsection + one API-reference block; added parity table with the asynclogparser standalone script. Verified end-to-end on a 128MB production .log (4,262 records, 9 frames): analyze / iter / pandas / CSV / JSONL / parsed-text sinks all decode correctly against schemas fetched from Horizon-v2.
🟢 LOW — safe to fast-track · confidence 0.85This adds offline .log-reading helpers to the Python SDK without changing how existing decode APIs or the raw-MPLog CLI path work. New code lives in Top risks
Check before approving
8 tool(s) ran · DRS agentic v2 · #383 |
![semgrep-sast-meesho[bot]](https://avatars.githubusercontent.com/u/19436680?s=60&v=4){kind=small}
|
|
||
| [project.optional-dependencies] | ||
| gcs = [ | ||
| "google-cloud-storage>=2.0.0", |
There was a problem hiding this comment.
Semgrep identified a blocking 🔴 issue in your code:
Dependency "$MATCH" uses a range operator. Pin to exact version with == or use a lockfile (e.g. uv.lock, pdm.lock, poetry.lock). Range pins allow auto-upgrades to compromised versions in CI.
Why this might be safe to ignore:
This finding is in a project dependencies specification (pyproject.toml) where range operators like '>=' are standard practice for library compatibility. The rule message warns about supply chain attacks through auto-upgrades, but this requires a lockfile-based workflow (poetry.lock, pdm.lock, uv.lock) to actually pin versions in deployment, which is the correct approach for Python libraries that need to maintain compatibility ranges.
To resolve this comment:
🔧 No guidance has been designated for this issue. Fix according to your organization's approved methods.
💬 Ignore this finding
Reply with Semgrep commands to ignore this finding.
/fp <comment>for false positive/ar <comment>for acceptable risk/other <comment>for all other reasons
Alternatively, triage in Semgrep AppSec Platform to ignore the finding created by pyproject-dependency-range-pin.
You can view more details about this finding in the Semgrep AppSec Platform.
There was a problem hiding this comment.
🤖 Code Review — PR #383
Language: Python | Files reviewed: 5
| Severity | Count | |
|---|---|---|
| BLOCKER | 🚨 | 2 |
| SUGGESTION | 💡 | 4 |
| WARNING | 18 |
| ) from exc | ||
| working = zstd.ZstdDecompressor().decompress(working) | ||
|
|
||
| parsed = parse_mplog_protobuf(working) |
There was a problem hiding this comment.
🚨 BLOCKER — Catch protobuf parse failures per record in non-strict mode instead of aborting the entire file decode.
_decode_one_record calls parse_mplog_protobuf before any try/except, so one malformed payload raises out of iter_decoded_log_rows and stops every sink. The PR describes strict=False as warn-and-skip behavior, but this path crashes CSV/pandas/Spark decoding for the whole file.
| schema: Optional[List[FeatureInfo]] = None, | ||
| needed_columns: Optional[Collection[str]] = None, | ||
| strict: bool = False, | ||
| ): |
There was a problem hiding this comment.
Public function signatures in this package should be type hinted; callers and static analysis currently cannot see that this returns a pandas DataFrame.
| record_length = int.from_bytes(block[offset : offset + _LENGTH_PREFIX_SIZE], "little") | ||
| offset += _LENGTH_PREFIX_SIZE | ||
|
|
||
| if record_length == 0: |
There was a problem hiding this comment.
The documented/asynclogparser framing uses record_length==0 as a sentinel; continuing advances through padding and can interpret padding garbage as additional records, producing spurious payloads or parse failures.
| ``user_id``, ``tracking_id``, ``parent_entity``) and then the decoded | ||
| feature columns. | ||
| """ | ||
| rows: List[dict] = list( |
There was a problem hiding this comment.
decode_log_file materializes every decoded row on the driver before creating the Spark DataFrame; stream or parallelize the decode instead of list(...).
This Spark-facing API is documented as the sink for big files, but it first builds a Python list containing one dict per decoded entity, then sends that entire list to Spark for schema inference and DataFrame creation. At production log sizes this defeats Spark's distributed execution, drives high Python heap usage, and can OOM the driver before Spark gets any work. Prefer decoding in partitions/from files with an RDD or mapInPandas, or require users to use a streaming sink for single-node decode.
Also flagged on this line:
- 🚨 BLOCKER — Avoid materializing the entire .log decode into a driver-side list for the Spark sink.
|
|
||
| data_size = capacity - _HEADER_SIZE | ||
|
|
||
| if valid_data_bytes > capacity: |
There was a problem hiding this comment.
The frame body is capacity-8 bytes and valid_data_bytes describes bytes within that body; accepting values up to capacity allows malformed headers through and can make the parser reason about bytes that cannot exist in the body.
Same issue also flagged at:
py-sdk/inference_logging_client/inference_logging_client/log_reader.py:770
| else: | ||
| with open(args.input, "rb") as f: | ||
| data = f.read() | ||
| log_mode = _is_log_input(args.input) |
There was a problem hiding this comment.
_is_log_input treats any GCS URI as a framed .log file, so raw MPLog bytes stored in GCS cannot be decoded by the CLI and will be interpreted as a frame header. This is a behavior regression for users whose input path is remote but not a .log container.
| try: | ||
| import zstandard as zstd # type: ignore | ||
| working = zstd.ZstdDecompressor().decompress(working) | ||
| except ImportError: |
There was a problem hiding this comment.
Other decode paths raise a clear ImportError when compressed payloads need zstandard, but write_parsed_log catches ImportError and continues with still-compressed bytes. The resulting parsed log contains parse errors for every compressed record instead of failing clearly.
| ) | ||
| ) | ||
|
|
||
| if not rows: |
There was a problem hiding this comment.
💡 SUGGESTION — Preserve provided schema/needed_columns in empty Spark outputs instead of returning metadata-only columns.
When no rows decode, decode_log_file returns an empty DataFrame containing only entity_id and record metadata. If the caller supplied a schema or needed_columns, downstream jobs expecting those feature columns will fail even though the raw decode_mplog path creates empty DataFrames with feature columns.
|
|
||
| ordered_cols = ["entity_id", *_RECORD_METADATA_COLUMNS, *feature_names] | ||
| # Backfill so every row has every column (Spark needs uniform dict keys). | ||
| for row in rows: |
There was a problem hiding this comment.
💡 SUGGESTION — Backfilling every row against every column adds an O(rows × columns) Python pass after all rows have already been decoded.
For sparse or wide feature sets, this nested loop performs a dict lookup/set for every row-column pair before createDataFrame. Spark can accept rows with missing dict keys as nulls during schema-driven creation, or the code can construct rows directly in the target column order to avoid this extra full pass.
| ) from exc | ||
|
|
||
| bucket_name, blob_name = _parse_gcs_uri(source) # type: ignore[arg-type] | ||
| client = storage.Client() |
There was a problem hiding this comment.
💡 SUGGESTION — Opening every gs:// source constructs a new storage.Client; allow client reuse for batch log processing.
storage.Client() performs credential setup and owns HTTP transport state, so creating it once per file prevents connection/client reuse. The README's partition fan-out example decodes many blobs in a loop; accepting an injected client or caching a module-level client avoids repeated setup overhead.
…_logs Adds decode_logs, decode_logs_to_pandas, and list_log_sources so the same API accepts a single .log file, a local directory, a gs:// prefix, or an explicit list of sources — and returns one merged DataFrame. - log_reader.py: new list_log_sources that dispatches on path shape (single file / local dir / GCS URI / prefix / list); decode_logs and decode_logs_to_pandas union across every listed source with unionByName(allowMissingColumns=True) / pd.concat. - __init__.py: export the three new public functions. - cli.py: default log-mode Spark path now calls decode_logs so directories and prefixes work; --output-format csv/jsonl/text on a multi-source input writes one file per input into --output as a directory; --output-format analyze on a multi-source input inspects the first file (with a note). - readme.md: adds a "Single file vs whole hour partition" subsection and extends the sinks table with the three new functions. Verified on gs://gcs-dsci-inferflow-async-logger-prd/.../2026-06-30/23/ (hour-level prefix): list_log_sources found 130 .log files; decode_logs_to_pandas on a 3-file subset produced a 26,364 x 550 merged DataFrame; CLI --output-format analyze on the bare prefix worked.
There was a problem hiding this comment.
🤖 Code Review — PR #383
Language: Python | Files reviewed: 5
| Severity | Count | |
|---|---|---|
| SUGGESTION | 💡 | 2 |
| WARNING | 11 |
|
|
||
| from functools import reduce | ||
|
|
||
| dfs = [ |
There was a problem hiding this comment.
decode_logs decodes every source eagerly and sequentially on the driver before returning the unioned Spark DataFrame.
The list comprehension calls decode_log_file for each source immediately, and each call itself materializes that file on the driver. This removes Spark parallelism across files and makes prefix-scale decoding bounded by one Python process and driver heap. Build a distributed input over the source list and decode in partitions, or at least lazily union without holding all per-file work in a single eager list.
| if not sources: | ||
| return pd.DataFrame(columns=["entity_id", *_RECORD_METADATA_COLUMNS]) | ||
|
|
||
| frames = [ |
There was a problem hiding this comment.
decode_logs_to_pandas should not keep every per-file DataFrame in memory before concatenation for multi-source inputs.
The pandas multi-file path builds a list of all decoded DataFrames and then pd.concat allocates another combined DataFrame. This doubles peak memory for large prefixes and delays output until all files have decoded. Concatenate in bounded batches or expose a streaming/chunked iterator for multi-source pandas use.
| ) | ||
| return entries | ||
|
|
||
| if os.path.isfile(source): |
There was a problem hiding this comment.
The function documentation says it enumerates .log files, but any existing file is returned, causing decode_logs('raw.bin') or decode_logs('notes.txt') to attempt .log deframing instead of rejecting or returning no sources.
| bucket_name, prefix = rest, "" | ||
| client = storage.Client() | ||
| blobs = client.list_blobs(bucket_name, prefix=prefix) | ||
| found = sorted( |
There was a problem hiding this comment.
sorted(...) consumes every matching blob under the prefix into a list before returning anything. For day/month prefixes this delays first-byte processing and stores all object names in memory even though decoding could be streamed page-by-page. Return an iterator or make sorting optional for very large prefixes.
| total = 0 | ||
| ext = {"csv": ".csv", "jsonl": ".jsonl", "text": ".parsed.log"}[out_fmt] | ||
| for s in sources: | ||
| base = os.path.basename(s.rstrip("/")).rsplit(".log", 1)[0] + ext |
There was a problem hiding this comment.
Using only os.path.basename causes files with the same name in different directories or GCS prefixes to overwrite each other in the output directory.
Also flagged on this line:
⚠️ WARNING — Multi-source CLI output can overwrite files when different source paths share the same basename.
| "MPLog payload is zstd-compressed but the 'zstandard' package is not " | ||
| "installed. Install it with: pip install zstandard" | ||
| ) from exc | ||
| working = zstd.ZstdDecompressor().decompress(working) |
There was a problem hiding this comment.
The decompressor call is not caught, so zstd errors propagate out of _decode_one_record. This is inconsistent with the default warn-and-skip behavior for malformed log data and can stop processing of otherwise usable records.
|
|
||
| out.write(f"=== Record {record_count} ===\n") | ||
| if ts_ns: | ||
| ts = _dt.datetime.fromtimestamp(ts_ns / 1e9, tz=_dt.timezone.utc) |
There was a problem hiding this comment.
write_parsed_log can crash on malformed or old-format records because timestamp conversion is not guarded.
The deframer always treats the first 8 record bytes as a Unix-nanosecond timestamp and passes it directly to datetime.fromtimestamp. If a record lacks the timestamp prefix or contains a bogus timestamp, this conversion can raise and abort the entire parsed-log write.
| "frames": len(frames), | ||
| "valid_bytes": total_valid, | ||
| "records": total_records, | ||
| "utilization": (total_valid / file_bytes) if file_bytes else None, |
There was a problem hiding this comment.
analyze_log_file computes utilization using valid bytes divided by total file bytes, underreporting utilization by including every frame header in the denominator.
total_valid sums only the valid body bytes, while file_bytes includes 8 bytes of header per frame and padding. For structural diagnosis this reports a lower utilization than the actual valid-data share of frame bodies, which can mislead capacity analysis.
| pdf.to_csv(args.output, index=False) | ||
| print(f"Output written to {args.output} ({len(pdf)} rows)") | ||
| else: | ||
| print(pdf.to_csv(index=False)) |
There was a problem hiding this comment.
💡 SUGGESTION — Stream pandas CSV output to stdout instead of building one giant CSV string with to_csv(index=False).
When no output path is supplied, print(pdf.to_csv(index=False)) allocates the entire CSV representation as a Python string in addition to the already-materialized DataFrame. For large .log files this can double memory and stall until formatting completes. Use pdf.to_csv(sys.stdout, index=False) or require --output above a safe row threshold.
| ] | ||
| if len(dfs) == 1: | ||
| return dfs[0] | ||
| return reduce(lambda a, b: a.unionByName(b, allowMissingColumns=True), dfs) |
There was a problem hiding this comment.
💡 SUGGESTION — Avoid building a left-deep unionByName chain for large source lists.
Reducing hundreds or thousands of DataFrames with pairwise unionByName creates a very large logical plan and can make Spark analysis/planning time significant before execution. For many files, decode them into one DataFrame from a single distributed source or periodically checkpoint/repartition batches before unioning. This keeps the plan size bounded as file count grows.
1 participant
Summary
.logreader toinference-logging-client. Accepts local paths, open binary file-like objects, andgs://bucket/keyURIs (via a new optional[gcs]extra)..parsed.logparity), and a structuralanalyzereport — all backed by one shared decode path (iter_decoded_log_rows) so row semantics are identical across sinks..log/gs://inputs and gains--output-format {spark,pandas,csv,jsonl,text,analyze}+--strict..logdocumentation is now one Quick Start subsection + one API-reference block, with a parity table vs the standaloneasynclogparserscript.What's new
decode_log_file(source, spark, ...)decode_log_file_to_pandas(source, ...)decode_log_file_to_csv(source, path, ...)decode_log_file_to_jsonl(source, path, ...)write_parsed_log(source, path, ...).parsed.logtextanalyze_log_file(source)+print_analysis(...)iter_decoded_log_rows(source, ...)iter_log_records(source, strict=False)(ts_ns, mplog_bytes)read_log_file(source, strict=False)All decoding functions share the same kwargs:
inference_host,decompress,schema=(pre-fetched),needed_columns=,strict=.Parity with the standalone
asynclogparserscriptThe byte-level deframer logic matches: 8-byte frame header →
capacity−8body → 4-byte length-prefix loop →record_length==0sentinel →[8B ts][MPLog payload]extraction. Additions the SDK carries: zstd decompression of MPLog payloads, cross-file schema cache keyed by(mp_config_id, version), andstrict=Truefail-fast for CI. The ~250 lines of byte-scan recovery heuristics from the script are intentionally not carried —strict=False(the default) warns-and-skips instead.Test plan
pip install dist/inference_logging_client-0.3.1-py3-none-any.whl, import fromsite-packages/verified (not source-tree fallback).analyze_log_fileon 128 MB prod log → 9 frames, 4,262 records, timestamp prefix detected.iter_log_records+parse_mplog_protobuf→ timestamps land in expected UTC range,mp_config_idmatches filename, 224-240 entities/record.get_feature_schemaagainst Horizon-v2 → 16 features returned.iter_decoded_log_rowswith pre-fetched schema → 500 rows decoded with real feature values.decode_log_file_to_csvon first frame → 95,086 rows / 23 MB.decode_log_file_to_jsonlon first frame → 95,086 rows / 82 MB.write_parsed_logon first frame → 525 records / 61 MB, byte-for-byte layout match with asynclogparser.inference-logging-client --output-format analyze <file>from the installed CLI script.FormatErroron corrupt frame; strict=False warns and continues.Notes for reviewers
decode_mplog/decode_mplog_dataframebehaviour or signatures.google-cloud-storageis an optional dep behind[gcs]— non-GCS users pay nothing.0.3.1inpyproject.toml; happy to bump to0.4.0before publish since this is a feature-adding release.🤖 Generated with Claude Code