PQ could benefit from event-level compression

[yaauie]

As write throughput is bound by disk IO, compressing events during serialization could improve throughput at the cost of CPU (see: proof of concept).

If possible, per-event compression should be delivered inside the scope of existing v2 PQ page format, in which entries contain only seqnum+length+N bytes. To do this, the reader will need to be able to handle compressed or uncompressed bytes without additional context (e.g., by differentiating zlib header from existing CBOR first-bytes).

Because not all users will want to spend CPU for increased throughput, and because of later-mentioned rollback barriers, this feature should first be delivered as opt-in, preferably at a per-pipeline level.

Compatibility Considerations

Once a queue contains compressed events, it will be unable to be read by a logstash instance that does not support event decompression; this presents an undesired rollback barrier that would prevent a user from rolling back to a last known-working configuration due to an unrelated issue.

Queue compression should be implemented as opt-in until at least three minor versions have shipped with decompression support.

Design Requirements

1. compression is opt-in for at least 2 minor releases
2. reads compressed events from queue unless explicitly configured otherwise
3. include metrics in pipeline.${pipeline_id}.queue.compression namespace:

   name	definition	expected value range
   encode.spend.lifetime	(encode_time / uptime)	[0,(N_CPUS)]
   encode.ratio.lifetime	(compressed_bytes / decompresssed_bytes)	[0,1]
   decode.spend.lifetime	(decode_time / uptime)	[0,(N_CPUS)]
   decode.ratio.lifetime	(decompressed_bytes / compresssed_bytes)	[1,)

[yaauie]

EDIT: The implementation in #18121 uses zstd instead of zlib deflate; the below is preserved for posterity but not relevant to the feature as-delivered.

Notes on disambiguating deflate from existing uncompressed CBOR payloads

Per RFC-1950, an example deflate-encoded value begins with:

+========CMF=======+=============FLG=============+
| +-CINFO-+--CM--+ | +-FLEVEL-+-FDICT-+-FCHECK-+ |
| | 0XXX  | 1000 | | | XX     | 0     | XXXXX  | |
| +-------+------+ | +--------+-------+--------+ |
+==================+=============================+

In practice, FDICT is always 0 (indicating no pre-determined dictionary was supplied).

The FLEVEL depends on the compression level, and FCHECK is a checksum ensuring that the two-byte sequence CMF+FLG interpreted as a MSB unsigned int is a multiple of 31.

Additionally, CINFO is usually 0111 (indicating the default 32KB window size), and is never 1XXX (since values above 7 are not allowed).

+========CMF=======+=============FLG=============+
| +-CINFO-+--CM--+ | +-FLEVEL-+-FDICT-+-FCHECK-+ |
| | 0111  | 1000 | | | XX     | 0     | XXXXX  | |
| +-------+------+ | +--------+-------+--------+ |
+==================+=============================+

Taken together, we can reliably determine that the payload is deflate-encoded if the first two bits satisfy:

• deflate: (bits & b'0F00' ^ b'0800') => b'0000'
• f_dict: (bits & b'0020') => b'0000'
• f_check: (bits % b'001f') => b'0000'
• c_info (strong): (bits & b'8000') => b'0000'
• c_info (weak): (bits & b'7000' ^ b'7000') => b'0000'
• deflate + c_info (strong) + f_dict: (bits & b'8F20' ^ b'0800') => b'0000'

---

Since at least 6.8, each event has always begin with the sequence:

10011111 01110001 01101010 01100001 01110110 01100001 00101110 01110101 01110100 01101001
01101100 00101110 01001000 01100001 01110011 01101000 01001101 01100001 01110000 10111111

9f                                                               # array(*)
   71                                                            #   text(17)
      6a6176612e7574696c2e486173684d6170                         #     "java.util.HashMap"
   bf                                                            #   map(*)

Additionally, with #17849 proposal to add the stringref extension, events would begin with D90100; or if we enable the CBOR self-describing tag D9D9F7.

bytes	description	deflate	f_dict	f_check	c_info (strong)	c_info (weak)
7801	deflate low/none	0000✅	0000✅	0000✅	0000✅	0000✅
785E	deflate fast	0000✅	0000✅	0000✅	0000✅	0000✅
789C	deflate default	0000✅	0000✅	0000✅	0000✅	0000✅
78DA	deflate best	0000✅	0000✅	0000✅	0000✅	0000✅
9F71	existing CBOR	0700❌	0020❌	0015❌	8000❌	6000❌
D819	stringref CBOR	0000✅	0000✅	0011❌	8000❌	2000❌
D9D9	self-described CBOR	0100❌	0000✅	0000✅	8000❌	2000❌

Therefore: any byte sequence that begins with a valid zlib deflate header is NOT a CBOR that we could have generated in the past, nor is it one that could be generated with the extensions under current consideration.

Additionally, because CBOR encodes the major type into the first three bits of the first byte, and because a deflate header's first bit is always 0 (due to its C_INFO constraint), a CBOR stream whose first element has a major-type of ARRAY(4) (100 XXXXX), MAP(5) (101 XXXXX), TAG(6) (110 XXXXX), or SIMPLE/FLOAT(7) (111 XXXXX) has a bit set that precludes it from being valid c_info for deflate-encoded streams.

This does NOT hold for types UINT(0) (000 XXXXX), NINT(1) (001 XXXXX), BYTE(2) (010 XXXXX) or TEXT(3) (011 XXXXX). For example, the common valid zlib deflate headers 7801, 785E, 789C, and 78DA are the CBOR encoding representing a text field with length of 1, 94, 156, and 218-bytes in length respectively :

78 01 # text(1)
   21 #   "!"

Conclusion

1. We should activate zlib-deflate decoding if the 2-byte header looks like zlib, and should fast-fail in order:
   • c_info (strong), f_dict, and deflate tests can be done in the same bitwise operation chain with ((bits & b'8F20') ^ b'0800') => b'0000'; the first operation zeroes out bits we don't care about, and the second flips the bits that we hope to be 1, so that we succeed if the result is fully zero-ed out.
   • f_check (bits % b'001f') => b'0000': slightly more expensive operation to check the checksum
2. We should harden the existing non-compressed CBOR to ensure it never looks like a valid zlib deflate header by either adding the stringref extension (which compacts the output) OR by enabling the 3-byte self-describing header.

[yaauie]

Notes on Configuration

In order to avoid corrupting a queue by changing the compression configuration, a queue should always be able to read both compressed and non-compressed events, regardless of the current configuration.

• queue.compression.enabled:
  • false (default for 2+ minor releases): does NOT compress when writing, but reads compressed events
  • true: compresses events when writing

We may want to expose the following configuration:

• queue.compression.method: (warn unless queue.compression.enabled == true)
  • deflate (default)
• queue.compression.deflate.level: (error unless queue.compression.method == deflate)
  • default (default) => effectively 6
  • [0,9]
  • fast => 1
  • best => 9

[EDIT]

For simplicity of configuration, this can be reduced to a single configuration if we avoid leaking the deflate-specific details at the config level:

• queue.compression
  • none: default; reads compressed events, but does not perform compression
  • speed: spend minimal CPU to get some compression benefits
  • balanced: spend more CPU for more compression
  • size: spend most CPU for smallest-possible
  • disabled: expert-mode opt-out, disables compression/decompression support entirely

[/EDIT]

I would prefer to make these configurable at the pipeline level, but doing so entails making these configurations available to the queue, and routing the queue's calls to Queueable#serialize and Queue#deserialize through a configured middleware.

[yaauie]

Persisted Queue vs Dead Letter Queue

This feature should be limited to the Persisted Queue, and should not affect the Dead Letter Queue, partially because of the complexity surrounding page boundaries in the DLQ's tests. If and when we wish to apply this to the DLQ, it will be done in a separate effort with separate considerations.

[robbavey]

@yaauie Can this be closed now?