Files
+6 835e1fd250 [rhythm] Introduce block-builder and kafka ingest path (#4533)
* Block-builder PoC

* Add unit test for block-builder (#4289)

* Add unit test for block-builder

* fmt

* Update tests

* cmon

* Deterministically build blocks for partition sections (#4327)

* Pull main (#4342)

* chore: remove gofakeit dependency (#4274)

* Further reduce Labes() calls in the metrics registry (#4283)

* Respect passed headers in read path requests (#4287)

* Ingester: Validate completed blocks (#4256)

* Add validate method to block

Signed-off-by: Joe Elliott <[email protected]>

* Add Validate usage in the ingester

Signed-off-by: Joe Elliott <[email protected]>

* changelog

Signed-off-by: Joe Elliott <[email protected]>

* add test and fix replay

Signed-off-by: Joe Elliott <[email protected]>

* increment metric

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>

* Add `invalid_utf8` to reasons spans could be rejected (#4293)

* Add `invalid_utf8` to reasons spans could be rejected

* Update changelog

* Update docs

* Ensure test covers invalid UTF-8 and not slack time

* add signals for duplicate rf1 data (#4296)

Signed-off-by: Joe Elliott <[email protected]>

* Bump anchore/sbom-action from 0.17.5 to 0.17.7 (#4307)

Bumps [anchore/sbom-action](https://github.com/anchore/sbom-action) from 0.17.5 to 0.17.7.
- [Release notes](https://github.com/anchore/sbom-action/releases)
- [Changelog](https://github.com/anchore/sbom-action/blob/main/RELEASE.md)
- [Commits](https://github.com/anchore/sbom-action/compare/v0.17.5...v0.17.7)

---
updated-dependencies:
- dependency-name: anchore/sbom-action
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* docs: Update readme with explore traces info (#4263)

* docs: Update readme with explore traces info

Co-authored-by: Kim Nylander <[email protected]>

* chore: remove spanlogger (#4312)

* chore: remove spanlogger

* Query-Frontend: Add middleware to drop headers (#4298)

* header strip ware

Signed-off-by: Joe Elliott <[email protected]>

* comment

Signed-off-by: Joe Elliott <[email protected]>

* changelog

Signed-off-by: Joe Elliott <[email protected]>

* remove header strip wear from metrics summary

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>

* Increase length of time compactions have to fail (#4315)

* increase length of time compactions have to fail

Signed-off-by: Joe Elliott <[email protected]>

* gen

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>

* docs: mark serverless as deprecated (#4017)

* docs: mark serverless as deprecated

* Changelog + readme

* docs: Remove duplicated examples (#4295)

This removes duplicates examples from the Configure TraceQL
metrics page.

Signed-off-by: Alex Bikfalvi <[email protected]>

* tempo-cli: support dropping multiple traces in a single operation (#4266)

* tempo-cli: support dropping multiple traces in a single operation

* update final log message

---------

Co-authored-by: Suraj Nath <[email protected]>

* [DOC] Add clarification for metrics summary and traceQL metrics (#4316)

* Add clarification for metrics summary and traceQL metrics

* Apply suggestions from code review

Co-authored-by: Jennifer Villa <[email protected]>

* Update docs/sources/tempo/api_docs/metrics-summary.md

---------

Co-authored-by: Jennifer Villa <[email protected]>

* TraceQL metrics time range fixes (#4325)

* Disconnect job time range filtering from step, so that results in split backend/recent range is accurate

* changelog

* Fix to assert metrics query range before alignment because alignment may increase it, which is not the responsibility of the caller to account for (#4331)

* Add doc about configuring TLS with Helm (#4328)

* Add doc about configuring TLS with Helm

* Add memberlist and readinessProbe to example

* Include server config for listening on TLS

* Add note about scraping

* Update docs/sources/tempo/configuration/network/tls.md

Co-authored-by: Markus Toivonen <[email protected]>

* Update docs/sources/tempo/configuration/network/tls.md

Co-authored-by: Kim Nylander <[email protected]>

* Update docs/sources/tempo/configuration/network/tls.md

Co-authored-by: Kim Nylander <[email protected]>

* Add memcached config for TLS

---------

Co-authored-by: Markus Toivonen <[email protected]>
Co-authored-by: Kim Nylander <[email protected]>

* [DOC] Add TLS info to Helm chart doc (#4334)

---------

Signed-off-by: Joe Elliott <[email protected]>
Signed-off-by: dependabot[bot] <[email protected]>
Signed-off-by: Alex Bikfalvi <[email protected]>
Co-authored-by: Javier Molina Reyes <[email protected]>
Co-authored-by: Zach Leslie <[email protected]>
Co-authored-by: Joe Elliott <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Ryan Perry <[email protected]>
Co-authored-by: Kim Nylander <[email protected]>
Co-authored-by: Suraj Nath <[email protected]>
Co-authored-by: Alex Bikfalvi <[email protected]>
Co-authored-by: Andrey Karpov <[email protected]>
Co-authored-by: Jennifer Villa <[email protected]>
Co-authored-by: Martin Disibio <[email protected]>
Co-authored-by: Markus Toivonen <[email protected]>

* WIP: Rhythm ingest path (#4314)

* Validate distributor config. Finish encoder/decoder tests

* Repair tests

* Make SingleBinary work out of the box by defaulting to partition 0

* Fix first time startup where blockbuilder fails before ingester can create topic

* Fix initial startup cycle time and delay

* Add more failure modes to the block-builder (#4345)

* Add more tests to the block-builder

* stuff

* Add comments

* [Rhythm] Metrics generator read from kafka first pass (#4359)

* Metrics generator read from kafka first pass

* review feedback

* Multiple fixes in block-builder (#4364)

* [rhythm] git merge origin/main (#4376)

* chore: remove gofakeit dependency (#4274)

* Further reduce Labes() calls in the metrics registry (#4283)

* Respect passed headers in read path requests (#4287)

* Ingester: Validate completed blocks (#4256)

* Add validate method to block

Signed-off-by: Joe Elliott <[email protected]>

* Add Validate usage in the ingester

Signed-off-by: Joe Elliott <[email protected]>

* changelog

Signed-off-by: Joe Elliott <[email protected]>

* add test and fix replay

Signed-off-by: Joe Elliott <[email protected]>

* increment metric

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>

* Add `invalid_utf8` to reasons spans could be rejected (#4293)

* Add `invalid_utf8` to reasons spans could be rejected

* Update changelog

* Update docs

* Ensure test covers invalid UTF-8 and not slack time

* add signals for duplicate rf1 data (#4296)

Signed-off-by: Joe Elliott <[email protected]>

* Bump anchore/sbom-action from 0.17.5 to 0.17.7 (#4307)

Bumps [anchore/sbom-action](https://github.com/anchore/sbom-action) from 0.17.5 to 0.17.7.
- [Release notes](https://github.com/anchore/sbom-action/releases)
- [Changelog](https://github.com/anchore/sbom-action/blob/main/RELEASE.md)
- [Commits](https://github.com/anchore/sbom-action/compare/v0.17.5...v0.17.7)

---
updated-dependencies:
- dependency-name: anchore/sbom-action
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* docs: Update readme with explore traces info (#4263)

* docs: Update readme with explore traces info

Co-authored-by: Kim Nylander <[email protected]>

* chore: remove spanlogger (#4312)

* chore: remove spanlogger

* Query-Frontend: Add middleware to drop headers (#4298)

* header strip ware

Signed-off-by: Joe Elliott <[email protected]>

* comment

Signed-off-by: Joe Elliott <[email protected]>

* changelog

Signed-off-by: Joe Elliott <[email protected]>

* remove header strip wear from metrics summary

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>

* Increase length of time compactions have to fail (#4315)

* increase length of time compactions have to fail

Signed-off-by: Joe Elliott <[email protected]>

* gen

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>

* docs: mark serverless as deprecated (#4017)

* docs: mark serverless as deprecated

* Changelog + readme

* docs: Remove duplicated examples (#4295)

This removes duplicates examples from the Configure TraceQL
metrics page.

Signed-off-by: Alex Bikfalvi <[email protected]>

* tempo-cli: support dropping multiple traces in a single operation (#4266)

* tempo-cli: support dropping multiple traces in a single operation

* update final log message

---------

Co-authored-by: Suraj Nath <[email protected]>

* [DOC] Add clarification for metrics summary and traceQL metrics (#4316)

* Add clarification for metrics summary and traceQL metrics

* Apply suggestions from code review

Co-authored-by: Jennifer Villa <[email protected]>

* Update docs/sources/tempo/api_docs/metrics-summary.md

---------

Co-authored-by: Jennifer Villa <[email protected]>

* TraceQL metrics time range fixes (#4325)

* Disconnect job time range filtering from step, so that results in split backend/recent range is accurate

* changelog

* Fix to assert metrics query range before alignment because alignment may increase it, which is not the responsibility of the caller to account for (#4331)

* Add doc about configuring TLS with Helm (#4328)

* Add doc about configuring TLS with Helm

* Add memberlist and readinessProbe to example

* Include server config for listening on TLS

* Add note about scraping

* Update docs/sources/tempo/configuration/network/tls.md

Co-authored-by: Markus Toivonen <[email protected]>

* Update docs/sources/tempo/configuration/network/tls.md

Co-authored-by: Kim Nylander <[email protected]>

* Update docs/sources/tempo/configuration/network/tls.md

Co-authored-by: Kim Nylander <[email protected]>

* Add memcached config for TLS

---------

Co-authored-by: Markus Toivonen <[email protected]>
Co-authored-by: Kim Nylander <[email protected]>

* [DOC] Add TLS info to Helm chart doc (#4334)

* fix deprecation warning by switching to DoBatchWithOptions (#4343)

Signed-off-by: Daniel Strobusch <[email protected]>

* bump dskit to v0.0.0-20241115082728-f2a7eb3aa0e9 to leverage benefits for context causes for DoBatch calls. (#4341)

See https://github.com/grafana/dskit/issues/576

Signed-off-by: Daniel Strobusch <[email protected]>

* Bump github.com/minio/minio-go/v7 from 7.0.70 to 7.0.80 (#4282)

* Bump github.com/minio/minio-go/v7 from 7.0.70 to 7.0.80

Bumps [github.com/minio/minio-go/v7](https://github.com/minio/minio-go) from 7.0.70 to 7.0.80.
- [Release notes](https://github.com/minio/minio-go/releases)
- [Commits](https://github.com/minio/minio-go/compare/v7.0.70...v7.0.80)

---
updated-dependencies:
- dependency-name: github.com/minio/minio-go/v7
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <[email protected]>

* Update serverless vendor

---------

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Zach Leslie <[email protected]>

* update default config values to better align with production workloads (#4340)

* update default config values to better align with production workloads

* Update CHANGELOG.md and config docs

* Ingester memory improvements by adjusting prealloc (#4344)

* remove trace ids

Signed-off-by: Joe Elliott <[email protected]>

* linear buckets

Signed-off-by: Joe Elliott <[email protected]>

* changelog

Signed-off-by: Joe Elliott <[email protected]>

* tuney tune

Signed-off-by: Joe Elliott <[email protected]>

* metric misses and increase pool size

Signed-off-by: Joe Elliott <[email protected]>

* lint

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>

* Bump github.com/Azure/azure-sdk-for-go/sdk/azcore from 1.13.0 to 1.16.0 (#4302)

* Bump github.com/Azure/azure-sdk-for-go/sdk/azcore from 1.13.0 to 1.16.0

Bumps [github.com/Azure/azure-sdk-for-go/sdk/azcore](https://github.com/Azure/azure-sdk-for-go) from 1.13.0 to 1.16.0.
- [Release notes](https://github.com/Azure/azure-sdk-for-go/releases)
- [Changelog](https://github.com/Azure/azure-sdk-for-go/blob/main/documentation/release.md)
- [Commits](https://github.com/Azure/azure-sdk-for-go/compare/sdk/azcore/v1.13.0...sdk/azcore/v1.16.0)

---
updated-dependencies:
- dependency-name: github.com/Azure/azure-sdk-for-go/sdk/azcore
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>

* Update serverless vendor

---------

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Zach Leslie <[email protected]>

* Use Prometheus fast regexp (#4329)

* basic integration

Signed-off-by: Joe Elliott <[email protected]>

* patch tests for new meaning

Signed-off-by: Joe Elliott <[email protected]>

* patch up more tests

Signed-off-by: Joe Elliott <[email protected]>

* add basic tests

Signed-off-by: Joe Elliott <[email protected]>

* changelog + docs

Signed-off-by: Joe Elliott <[email protected]>

* remove benches

Signed-off-by: Joe Elliott <[email protected]>

* Cleaned up + tests

Signed-off-by: Joe Elliott <[email protected]>

* comment

Signed-off-by: Joe Elliott <[email protected]>

* lint

Signed-off-by: Joe Elliott <[email protected]>

* Update docs/sources/tempo/traceql/_index.md

Co-authored-by: Kim Nylander <[email protected]>

* comment

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>
Co-authored-by: Kim Nylander <[email protected]>

* Fix broken link in service-graphs docs (#4351)

* Fix minor typo in TraceQL docs (#4356)

* Bump default memcached version (#4363)

* Exemplar fixes (#4366)

* Fix exemplars based on duration to convert to seconds, fix various other issues

* changelog

* fix: initialize histogram buckets to 0 to avoid them being downsampled (#4368)

* initialized histogram buckets to 0 to avoid them being downsampled

* Ingester/Generator Live trace cleanup (#4365)

* moved trace sizes somewhere shareable

Signed-off-by: Joe Elliott <[email protected]>

* use tracesizes in ingester

Signed-off-by: Joe Elliott <[email protected]>

* make tests work

Signed-off-by: Joe Elliott <[email protected]>

* trace bytes in generator

Signed-off-by: Joe Elliott <[email protected]>

* remove traceCount

Signed-off-by: Joe Elliott <[email protected]>

* live trace shenanigans

Signed-off-by: Joe Elliott <[email protected]>

* changelog

Signed-off-by: Joe Elliott <[email protected]>

* Update modules/generator/processor/localblocks/livetraces.go

Co-authored-by: Mario <[email protected]>

* Update modules/ingester/instance.go

Co-authored-by: Mario <[email protected]>

* Test cleanup. Add sz test, restore commented out and fix e2e

Signed-off-by: Joe Elliott <[email protected]>

* remove todo comment

Signed-off-by: Joe Elliott <[email protected]>

---------

Signed-off-by: Joe Elliott <[email protected]>
Co-authored-by: Mario <[email protected]>

* Bump anchore/sbom-action from 0.17.7 to 0.17.8 (#4371)

Bumps [anchore/sbom-action](https://github.com/anchore/sbom-action) from 0.17.7 to 0.17.8.
- [Release notes](https://github.com/anchore/sbom-action/releases)
- [Changelog](https://github.com/anchore/sbom-action/blob/main/RELEASE.md)
- [Commits](https://github.com/anchore/sbom-action/compare/v0.17.7...v0.17.8)

---
updated-dependencies:
- dependency-name: anchore/sbom-action
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* Update for IDs change

* Only run blockbuilder if ingest enabled

---------

Signed-off-by: Joe Elliott <[email protected]>
Signed-off-by: dependabot[bot] <[email protected]>
Signed-off-by: Alex Bikfalvi <[email protected]>
Signed-off-by: Daniel Strobusch <[email protected]>
Co-authored-by: Javier Molina Reyes <[email protected]>
Co-authored-by: Zach Leslie <[email protected]>
Co-authored-by: Joe Elliott <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Ryan Perry <[email protected]>
Co-authored-by: Kim Nylander <[email protected]>
Co-authored-by: Suraj Nath <[email protected]>
Co-authored-by: Alex Bikfalvi <[email protected]>
Co-authored-by: Andrey Karpov <[email protected]>
Co-authored-by: Jennifer Villa <[email protected]>
Co-authored-by: Martin Disibio <[email protected]>
Co-authored-by: Markus Toivonen <[email protected]>
Co-authored-by: Daniel Strobusch <[email protected]>
Co-authored-by: Carles Garcia <[email protected]>

* [rhythm] Changes to simplify operations (#4389)

* Use mapping for assigning partitions

* Use mapping for assigning partitions in the generator too

* Add support for SASL auth to kafka clients

* Add metrics to ingest (#4395)

* [rhythm] Extract block-builder into its own module (#4396)

* Extract block-builder into its own module

* Update /operations and examples

* No ephemeral storage

* No rolling strategy either

* fmt and compile

* Address review comment

* [rhythm] Correctly pass start/end time when appending a trace (#4410)

* Correctly pass start/end times

* Different code, same result

* [rhythm] Multiple fixes to block-builder consumption (#4413)

* Multiple fixes to cycle consumption

* fmt

* happy now?

* ups

* Rhythm: Separate non-flushing local blocks processor to store new queue data for reads (#4411)

* wip: separate non-flushing local blocks processor to store new queue data for reads

* Make real config for non-flushing local blocks processor, optional, validate wal config and use defaults if needed

* Fix defaulting of second WAL config

* [rhythm] Make ID generator more robust (#4416)

* Make ID generator more robust

* Simplify

* Update to c3ff5ba61

* Fix registering of kafka read client metrics (#4502)

* [rhythm] Make ID generator more robust (#4416) (#4507)

* Make ID generator more robust

* Simplify

* Removed references to Loki and Mimir (#4509)

Signed-off-by: Joe Elliott <[email protected]>

* [Rhythm] Block builder test updates (#4510)

* Make blockbuilder tests closer to real kafka and less implementation specific by always enabling support for consumer groups, call commit control func in order

* Verify last committed offset in each test

* hide test function

* lint

* lint

* [Rhythm] Block-builder consumption loop (#4480)

* Alternate block-builder consume

* Set timeout on PollFetches, reduce initial poll delay, update 1 test to work using real consumergroup functionality

* restore metrics

* Re-add original partition lag metric, polled in separate goroutine. Fix consume loop to only consume full-duration cycles for more determinism

* merge conflict

* Review feedback

* Review feedback

* Comment

* code cleanup, lint

* logs

* code cleanup

* lint

* Review feedback

* Remove missed lookback_on_no_commit config in e2e tests and regen manifest

* Review feedback

* Fix rewind to latest commit to init correctly, it didn't work in some clusters (#4532)

* [rhythm] merge main at 2c44177 (#4531)

* Fixes

* More fixes

---------

Signed-off-by: Joe Elliott <[email protected]>
Signed-off-by: dependabot[bot] <[email protected]>
Signed-off-by: Alex Bikfalvi <[email protected]>
Signed-off-by: Daniel Strobusch <[email protected]>
Co-authored-by: Javier Molina Reyes <[email protected]>
Co-authored-by: Zach Leslie <[email protected]>
Co-authored-by: Joe Elliott <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Ryan Perry <[email protected]>
Co-authored-by: Kim Nylander <[email protected]>
Co-authored-by: Suraj Nath <[email protected]>
Co-authored-by: Alex Bikfalvi <[email protected]>
Co-authored-by: Andrey Karpov <[email protected]>
Co-authored-by: Jennifer Villa <[email protected]>
Co-authored-by: Martin Disibio <[email protected]>
Co-authored-by: Markus Toivonen <[email protected]>
Co-authored-by: Daniel Strobusch <[email protected]>
Co-authored-by: Carles Garcia <[email protected]>
2025-01-10 16:05:42 +01:00

857 lines
21 KiB
Go

// Package kbin contains Kafka primitive reading and writing functions.
package kbin
import (
"encoding/binary"
"errors"
"math"
"math/bits"
"reflect"
"unsafe"
)
// This file contains primitive type encoding and decoding.
//
// The Reader helper can be used even when content runs out
// or an error is hit; all other number requests will return
// zero so a decode will basically no-op.
// ErrNotEnoughData is returned when a type could not fully decode
// from a slice because the slice did not have enough data.
var ErrNotEnoughData = errors.New("response did not contain enough data to be valid")
// AppendBool appends 1 for true or 0 for false to dst.
func AppendBool(dst []byte, v bool) []byte {
if v {
return append(dst, 1)
}
return append(dst, 0)
}
// AppendInt8 appends an int8 to dst.
func AppendInt8(dst []byte, i int8) []byte {
return append(dst, byte(i))
}
// AppendInt16 appends a big endian int16 to dst.
func AppendInt16(dst []byte, i int16) []byte {
return AppendUint16(dst, uint16(i))
}
// AppendUint16 appends a big endian uint16 to dst.
func AppendUint16(dst []byte, u uint16) []byte {
return append(dst, byte(u>>8), byte(u))
}
// AppendInt32 appends a big endian int32 to dst.
func AppendInt32(dst []byte, i int32) []byte {
return AppendUint32(dst, uint32(i))
}
// AppendInt64 appends a big endian int64 to dst.
func AppendInt64(dst []byte, i int64) []byte {
return appendUint64(dst, uint64(i))
}
// AppendFloat64 appends a big endian float64 to dst.
func AppendFloat64(dst []byte, f float64) []byte {
return appendUint64(dst, math.Float64bits(f))
}
// AppendUuid appends the 16 uuid bytes to dst.
func AppendUuid(dst []byte, uuid [16]byte) []byte {
return append(dst, uuid[:]...)
}
func appendUint64(dst []byte, u uint64) []byte {
return append(dst, byte(u>>56), byte(u>>48), byte(u>>40), byte(u>>32),
byte(u>>24), byte(u>>16), byte(u>>8), byte(u))
}
// AppendUint32 appends a big endian uint32 to dst.
func AppendUint32(dst []byte, u uint32) []byte {
return append(dst, byte(u>>24), byte(u>>16), byte(u>>8), byte(u))
}
// uvarintLens could only be length 65, but using 256 allows bounds check
// elimination on lookup.
const uvarintLens = "\x01\x01\x01\x01\x01\x01\x01\x01\x02\x02\x02\x02\x02\x02\x02\x03\x03\x03\x03\x03\x03\x03\x04\x04\x04\x04\x04\x04\x04\x05\x05\x05\x05\x05\x05\x05\x06\x06\x06\x06\x06\x06\x06\x07\x07\x07\x07\x07\x07\x07\x08\x08\x08\x08\x08\x08\x08\x09\x09\x09\x09\x09\x09\x09\x0a\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
// VarintLen returns how long i would be if it were varint encoded.
func VarintLen(i int32) int {
u := uint32(i)<<1 ^ uint32(i>>31)
return UvarintLen(u)
}
// UvarintLen returns how long u would be if it were uvarint encoded.
func UvarintLen(u uint32) int {
return int(uvarintLens[byte(bits.Len32(u))])
}
// VarlongLen returns how long i would be if it were varlong encoded.
func VarlongLen(i int64) int {
u := uint64(i)<<1 ^ uint64(i>>63)
return uvarlongLen(u)
}
func uvarlongLen(u uint64) int {
return int(uvarintLens[byte(bits.Len64(u))])
}
// Varint is a loop unrolled 32 bit varint decoder. The return semantics
// are the same as binary.Varint, with the added benefit that overflows
// in 5 byte encodings are handled rather than left to the user.
func Varint(in []byte) (int32, int) {
x, n := Uvarint(in)
return int32((x >> 1) ^ -(x & 1)), n
}
// Uvarint is a loop unrolled 32 bit uvarint decoder. The return semantics
// are the same as binary.Uvarint, with the added benefit that overflows
// in 5 byte encodings are handled rather than left to the user.
func Uvarint(in []byte) (uint32, int) {
var x uint32
var overflow int
if len(in) < 1 {
goto fail
}
x = uint32(in[0] & 0x7f)
if in[0]&0x80 == 0 {
return x, 1
} else if len(in) < 2 {
goto fail
}
x |= uint32(in[1]&0x7f) << 7
if in[1]&0x80 == 0 {
return x, 2
} else if len(in) < 3 {
goto fail
}
x |= uint32(in[2]&0x7f) << 14
if in[2]&0x80 == 0 {
return x, 3
} else if len(in) < 4 {
goto fail
}
x |= uint32(in[3]&0x7f) << 21
if in[3]&0x80 == 0 {
return x, 4
} else if len(in) < 5 {
goto fail
}
x |= uint32(in[4]) << 28
if in[4] <= 0x0f {
return x, 5
}
overflow = -5
fail:
return 0, overflow
}
// Varlong is a loop unrolled 64 bit varint decoder. The return semantics
// are the same as binary.Varint, with the added benefit that overflows
// in 10 byte encodings are handled rather than left to the user.
func Varlong(in []byte) (int64, int) {
x, n := uvarlong(in)
return int64((x >> 1) ^ -(x & 1)), n
}
func uvarlong(in []byte) (uint64, int) {
var x uint64
var overflow int
if len(in) < 1 {
goto fail
}
x = uint64(in[0] & 0x7f)
if in[0]&0x80 == 0 {
return x, 1
} else if len(in) < 2 {
goto fail
}
x |= uint64(in[1]&0x7f) << 7
if in[1]&0x80 == 0 {
return x, 2
} else if len(in) < 3 {
goto fail
}
x |= uint64(in[2]&0x7f) << 14
if in[2]&0x80 == 0 {
return x, 3
} else if len(in) < 4 {
goto fail
}
x |= uint64(in[3]&0x7f) << 21
if in[3]&0x80 == 0 {
return x, 4
} else if len(in) < 5 {
goto fail
}
x |= uint64(in[4]&0x7f) << 28
if in[4]&0x80 == 0 {
return x, 5
} else if len(in) < 6 {
goto fail
}
x |= uint64(in[5]&0x7f) << 35
if in[5]&0x80 == 0 {
return x, 6
} else if len(in) < 7 {
goto fail
}
x |= uint64(in[6]&0x7f) << 42
if in[6]&0x80 == 0 {
return x, 7
} else if len(in) < 8 {
goto fail
}
x |= uint64(in[7]&0x7f) << 49
if in[7]&0x80 == 0 {
return x, 8
} else if len(in) < 9 {
goto fail
}
x |= uint64(in[8]&0x7f) << 56
if in[8]&0x80 == 0 {
return x, 9
} else if len(in) < 10 {
goto fail
}
x |= uint64(in[9]) << 63
if in[9] <= 0x01 {
return x, 10
}
overflow = -10
fail:
return 0, overflow
}
// AppendVarint appends a varint encoded i to dst.
func AppendVarint(dst []byte, i int32) []byte {
return AppendUvarint(dst, uint32(i)<<1^uint32(i>>31))
}
// AppendUvarint appends a uvarint encoded u to dst.
func AppendUvarint(dst []byte, u uint32) []byte {
switch UvarintLen(u) {
case 5:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte((u>>21)&0x7f|0x80),
byte(u>>28))
case 4:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte(u>>21))
case 3:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte(u>>14))
case 2:
return append(dst,
byte(u&0x7f|0x80),
byte(u>>7))
case 1:
return append(dst, byte(u))
}
return dst
}
// AppendVarlong appends a varint encoded i to dst.
func AppendVarlong(dst []byte, i int64) []byte {
return appendUvarlong(dst, uint64(i)<<1^uint64(i>>63))
}
func appendUvarlong(dst []byte, u uint64) []byte {
switch uvarlongLen(u) {
case 10:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte((u>>21)&0x7f|0x80),
byte((u>>28)&0x7f|0x80),
byte((u>>35)&0x7f|0x80),
byte((u>>42)&0x7f|0x80),
byte((u>>49)&0x7f|0x80),
byte((u>>56)&0x7f|0x80),
byte(u>>63))
case 9:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte((u>>21)&0x7f|0x80),
byte((u>>28)&0x7f|0x80),
byte((u>>35)&0x7f|0x80),
byte((u>>42)&0x7f|0x80),
byte((u>>49)&0x7f|0x80),
byte(u>>56))
case 8:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte((u>>21)&0x7f|0x80),
byte((u>>28)&0x7f|0x80),
byte((u>>35)&0x7f|0x80),
byte((u>>42)&0x7f|0x80),
byte(u>>49))
case 7:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte((u>>21)&0x7f|0x80),
byte((u>>28)&0x7f|0x80),
byte((u>>35)&0x7f|0x80),
byte(u>>42))
case 6:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte((u>>21)&0x7f|0x80),
byte((u>>28)&0x7f|0x80),
byte(u>>35))
case 5:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte((u>>21)&0x7f|0x80),
byte(u>>28))
case 4:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte((u>>14)&0x7f|0x80),
byte(u>>21))
case 3:
return append(dst,
byte(u&0x7f|0x80),
byte((u>>7)&0x7f|0x80),
byte(u>>14))
case 2:
return append(dst,
byte(u&0x7f|0x80),
byte(u>>7))
case 1:
return append(dst, byte(u))
}
return dst
}
// AppendString appends a string to dst prefixed with its int16 length.
func AppendString(dst []byte, s string) []byte {
dst = AppendInt16(dst, int16(len(s)))
return append(dst, s...)
}
// AppendCompactString appends a string to dst prefixed with its uvarint length
// starting at 1; 0 is reserved for null, which compact strings are not
// (nullable compact ones are!). Thus, the length is the decoded uvarint - 1.
//
// For KIP-482.
func AppendCompactString(dst []byte, s string) []byte {
dst = AppendUvarint(dst, 1+uint32(len(s)))
return append(dst, s...)
}
// AppendNullableString appends potentially nil string to dst prefixed with its
// int16 length or int16(-1) if nil.
func AppendNullableString(dst []byte, s *string) []byte {
if s == nil {
return AppendInt16(dst, -1)
}
return AppendString(dst, *s)
}
// AppendCompactNullableString appends a potentially nil string to dst with its
// uvarint length starting at 1, with 0 indicating null. Thus, the length is
// the decoded uvarint - 1.
//
// For KIP-482.
func AppendCompactNullableString(dst []byte, s *string) []byte {
if s == nil {
return AppendUvarint(dst, 0)
}
return AppendCompactString(dst, *s)
}
// AppendBytes appends bytes to dst prefixed with its int32 length.
func AppendBytes(dst, b []byte) []byte {
dst = AppendInt32(dst, int32(len(b)))
return append(dst, b...)
}
// AppendCompactBytes appends bytes to dst prefixed with a its uvarint length
// starting at 1; 0 is reserved for null, which compact bytes are not (nullable
// compact ones are!). Thus, the length is the decoded uvarint - 1.
//
// For KIP-482.
func AppendCompactBytes(dst, b []byte) []byte {
dst = AppendUvarint(dst, 1+uint32(len(b)))
return append(dst, b...)
}
// AppendNullableBytes appends a potentially nil slice to dst prefixed with its
// int32 length or int32(-1) if nil.
func AppendNullableBytes(dst, b []byte) []byte {
if b == nil {
return AppendInt32(dst, -1)
}
return AppendBytes(dst, b)
}
// AppendCompactNullableBytes appends a potentially nil slice to dst with its
// uvarint length starting at 1, with 0 indicating null. Thus, the length is
// the decoded uvarint - 1.
//
// For KIP-482.
func AppendCompactNullableBytes(dst, b []byte) []byte {
if b == nil {
return AppendUvarint(dst, 0)
}
return AppendCompactBytes(dst, b)
}
// AppendVarintString appends a string to dst prefixed with its length encoded
// as a varint.
func AppendVarintString(dst []byte, s string) []byte {
dst = AppendVarint(dst, int32(len(s)))
return append(dst, s...)
}
// AppendVarintBytes appends a slice to dst prefixed with its length encoded as
// a varint.
func AppendVarintBytes(dst, b []byte) []byte {
if b == nil {
return AppendVarint(dst, -1)
}
dst = AppendVarint(dst, int32(len(b)))
return append(dst, b...)
}
// AppendArrayLen appends the length of an array as an int32 to dst.
func AppendArrayLen(dst []byte, l int) []byte {
return AppendInt32(dst, int32(l))
}
// AppendCompactArrayLen appends the length of an array as a uvarint to dst
// as the length + 1.
//
// For KIP-482.
func AppendCompactArrayLen(dst []byte, l int) []byte {
return AppendUvarint(dst, 1+uint32(l))
}
// AppendNullableArrayLen appends the length of an array as an int32 to dst,
// or -1 if isNil is true.
func AppendNullableArrayLen(dst []byte, l int, isNil bool) []byte {
if isNil {
return AppendInt32(dst, -1)
}
return AppendInt32(dst, int32(l))
}
// AppendCompactNullableArrayLen appends the length of an array as a uvarint to
// dst as the length + 1; if isNil is true, this appends 0 as a uvarint.
//
// For KIP-482.
func AppendCompactNullableArrayLen(dst []byte, l int, isNil bool) []byte {
if isNil {
return AppendUvarint(dst, 0)
}
return AppendUvarint(dst, 1+uint32(l))
}
// Reader is used to decode Kafka messages.
//
// For all functions on Reader, if the reader has been invalidated, functions
// return defaults (false, 0, nil, ""). Use Complete to detect if the reader
// was invalidated or if the reader has remaining data.
type Reader struct {
Src []byte
bad bool
}
// Bool returns a bool from the reader.
func (b *Reader) Bool() bool {
if len(b.Src) < 1 {
b.bad = true
b.Src = nil
return false
}
t := b.Src[0] != 0 // if '0', false
b.Src = b.Src[1:]
return t
}
// Int8 returns an int8 from the reader.
func (b *Reader) Int8() int8 {
if len(b.Src) < 1 {
b.bad = true
b.Src = nil
return 0
}
r := b.Src[0]
b.Src = b.Src[1:]
return int8(r)
}
// Int16 returns an int16 from the reader.
func (b *Reader) Int16() int16 {
if len(b.Src) < 2 {
b.bad = true
b.Src = nil
return 0
}
r := int16(binary.BigEndian.Uint16(b.Src))
b.Src = b.Src[2:]
return r
}
// Uint16 returns an uint16 from the reader.
func (b *Reader) Uint16() uint16 {
if len(b.Src) < 2 {
b.bad = true
b.Src = nil
return 0
}
r := binary.BigEndian.Uint16(b.Src)
b.Src = b.Src[2:]
return r
}
// Int32 returns an int32 from the reader.
func (b *Reader) Int32() int32 {
if len(b.Src) < 4 {
b.bad = true
b.Src = nil
return 0
}
r := int32(binary.BigEndian.Uint32(b.Src))
b.Src = b.Src[4:]
return r
}
// Int64 returns an int64 from the reader.
func (b *Reader) Int64() int64 {
return int64(b.readUint64())
}
// Uuid returns a uuid from the reader.
func (b *Reader) Uuid() [16]byte {
var r [16]byte
copy(r[:], b.Span(16))
return r
}
// Float64 returns a float64 from the reader.
func (b *Reader) Float64() float64 {
return math.Float64frombits(b.readUint64())
}
func (b *Reader) readUint64() uint64 {
if len(b.Src) < 8 {
b.bad = true
b.Src = nil
return 0
}
r := binary.BigEndian.Uint64(b.Src)
b.Src = b.Src[8:]
return r
}
// Uint32 returns a uint32 from the reader.
func (b *Reader) Uint32() uint32 {
if len(b.Src) < 4 {
b.bad = true
b.Src = nil
return 0
}
r := binary.BigEndian.Uint32(b.Src)
b.Src = b.Src[4:]
return r
}
// Varint returns a varint int32 from the reader.
func (b *Reader) Varint() int32 {
val, n := Varint(b.Src)
if n <= 0 {
b.bad = true
b.Src = nil
return 0
}
b.Src = b.Src[n:]
return val
}
// Varlong returns a varlong int64 from the reader.
func (b *Reader) Varlong() int64 {
val, n := Varlong(b.Src)
if n <= 0 {
b.bad = true
b.Src = nil
return 0
}
b.Src = b.Src[n:]
return val
}
// Uvarint returns a uvarint encoded uint32 from the reader.
func (b *Reader) Uvarint() uint32 {
val, n := Uvarint(b.Src)
if n <= 0 {
b.bad = true
b.Src = nil
return 0
}
b.Src = b.Src[n:]
return val
}
// Span returns l bytes from the reader.
func (b *Reader) Span(l int) []byte {
if len(b.Src) < l || l < 0 {
b.bad = true
b.Src = nil
return nil
}
r := b.Src[:l:l]
b.Src = b.Src[l:]
return r
}
// UnsafeString returns a Kafka string from the reader without allocating using
// the unsafe package. This must be used with care; note the string holds a
// reference to the original slice.
func (b *Reader) UnsafeString() string {
l := b.Int16()
return UnsafeString(b.Span(int(l)))
}
// String returns a Kafka string from the reader.
func (b *Reader) String() string {
l := b.Int16()
return string(b.Span(int(l)))
}
// UnsafeCompactString returns a Kafka compact string from the reader without
// allocating using the unsafe package. This must be used with care; note the
// string holds a reference to the original slice.
func (b *Reader) UnsafeCompactString() string {
l := int(b.Uvarint()) - 1
return UnsafeString(b.Span(l))
}
// CompactString returns a Kafka compact string from the reader.
func (b *Reader) CompactString() string {
l := int(b.Uvarint()) - 1
return string(b.Span(l))
}
// UnsafeNullableString returns a Kafka nullable string from the reader without
// allocating using the unsafe package. This must be used with care; note the
// string holds a reference to the original slice.
func (b *Reader) UnsafeNullableString() *string {
l := b.Int16()
if l < 0 {
return nil
}
s := UnsafeString(b.Span(int(l)))
return &s
}
// NullableString returns a Kafka nullable string from the reader.
func (b *Reader) NullableString() *string {
l := b.Int16()
if l < 0 {
return nil
}
s := string(b.Span(int(l)))
return &s
}
// UnsafeCompactNullableString returns a Kafka compact nullable string from the
// reader without allocating using the unsafe package. This must be used with
// care; note the string holds a reference to the original slice.
func (b *Reader) UnsafeCompactNullableString() *string {
l := int(b.Uvarint()) - 1
if l < 0 {
return nil
}
s := UnsafeString(b.Span(l))
return &s
}
// CompactNullableString returns a Kafka compact nullable string from the
// reader.
func (b *Reader) CompactNullableString() *string {
l := int(b.Uvarint()) - 1
if l < 0 {
return nil
}
s := string(b.Span(l))
return &s
}
// Bytes returns a Kafka byte array from the reader.
//
// This never returns nil.
func (b *Reader) Bytes() []byte {
l := b.Int32()
// This is not to spec, but it is not clearly documented and Microsoft
// EventHubs fails here. -1 means null, which should throw an
// exception. EventHubs uses -1 to mean "does not exist" on some
// non-nullable fields.
//
// Until EventHubs is fixed, we return an empty byte slice for null.
if l == -1 {
return []byte{}
}
return b.Span(int(l))
}
// CompactBytes returns a Kafka compact byte array from the reader.
//
// This never returns nil.
func (b *Reader) CompactBytes() []byte {
l := int(b.Uvarint()) - 1
if l == -1 { // same as above: -1 should not be allowed here
return []byte{}
}
return b.Span(l)
}
// NullableBytes returns a Kafka nullable byte array from the reader, returning
// nil as appropriate.
func (b *Reader) NullableBytes() []byte {
l := b.Int32()
if l < 0 {
return nil
}
r := b.Span(int(l))
return r
}
// CompactNullableBytes returns a Kafka compact nullable byte array from the
// reader, returning nil as appropriate.
func (b *Reader) CompactNullableBytes() []byte {
l := int(b.Uvarint()) - 1
if l < 0 {
return nil
}
r := b.Span(l)
return r
}
// ArrayLen returns a Kafka array length from the reader.
func (b *Reader) ArrayLen() int32 {
r := b.Int32()
// The min size of a Kafka type is a byte, so if we do not have
// at least the array length of bytes left, it is bad.
if len(b.Src) < int(r) {
b.bad = true
b.Src = nil
return 0
}
return r
}
// VarintArrayLen returns a Kafka array length from the reader.
func (b *Reader) VarintArrayLen() int32 {
r := b.Varint()
// The min size of a Kafka type is a byte, so if we do not have
// at least the array length of bytes left, it is bad.
if len(b.Src) < int(r) {
b.bad = true
b.Src = nil
return 0
}
return r
}
// CompactArrayLen returns a Kafka compact array length from the reader.
func (b *Reader) CompactArrayLen() int32 {
r := int32(b.Uvarint()) - 1
// The min size of a Kafka type is a byte, so if we do not have
// at least the array length of bytes left, it is bad.
if len(b.Src) < int(r) {
b.bad = true
b.Src = nil
return 0
}
return r
}
// VarintBytes returns a Kafka encoded varint array from the reader, returning
// nil as appropriate.
func (b *Reader) VarintBytes() []byte {
l := b.Varint()
if l < 0 {
return nil
}
return b.Span(int(l))
}
// UnsafeVarintString returns a Kafka encoded varint string from the reader
// without allocating using the unsafe package. This must be used with care;
// note the string holds a reference to the original slice.
func (b *Reader) UnsafeVarintString() string {
return UnsafeString(b.VarintBytes())
}
// VarintString returns a Kafka encoded varint string from the reader.
func (b *Reader) VarintString() string {
return string(b.VarintBytes())
}
// Complete returns ErrNotEnoughData if the source ran out while decoding.
func (b *Reader) Complete() error {
if b.bad {
return ErrNotEnoughData
}
return nil
}
// Ok returns true if the reader is still ok.
func (b *Reader) Ok() bool {
return !b.bad
}
// UnsafeString returns the slice as a string using unsafe rule (6).
func UnsafeString(slice []byte) string {
var str string
strhdr := (*reflect.StringHeader)(unsafe.Pointer(&str)) //nolint:gosec // known way to convert slice to string
strhdr.Data = ((*reflect.SliceHeader)(unsafe.Pointer(&slice))).Data //nolint:gosec // known way to convert slice to string
strhdr.Len = len(slice)
return str
}