ClickHouse as a storage backend

[sboisson]

ClickHouse, an open-source column-oriented DBMS designed initially for real-time analytics and mostly write-once/read-many big-data use cases, can be used as a very efficient log and trace storage.

Meta issue: #638 Additional storage backends

[bzon]

Is anyone working on this? I or my team can maybe give a shot at this.

[Slach]

as i now noone working on this

[Slach]

@bzon i can join you as tester

[bzon]

@bzon i can join you as tester

Sure!

[sboisson]

Happy to see someone working on this :)
Might be able to join as tester

[yurishkuro]

@bzon would be good if you post an architecture here, specifically how you would lay out the data, ingestion, etc. To my knowledge, clickhouse requires batched writes, and it may even be up to you to decide which node to send the writes to, so there are many questions. It may require some benchmarking to find the optimal design.

[bzon]

@yurishkuro at the moment, we have zero knowledge with the internals of jaeger. I think that benchmarking should be the first step to see if this integration is feasible. And with that said, the first requirement should be creating the right table schema.

[sboisson]

I think architecture of project https://github.com/flant/loghouse could be a source of inspiration…

[sboisson]

This webinar could be interesting: A Practical Introduction to Handling Log Data in ClickHouse

[bobrik]

I took a stab at it (very early WIP):

• master...bobrik:ivan/clickhouse

This is the schema I used:

• Index table for fast searches (haven't measured if indices are useful yet)

CREATE TABLE jaeger_index (
  timestamp DateTime64(6),
  traceID FixedString(16),
  service LowCardinality(String),
  operation LowCardinality(String),
  durationUs UInt64,
  tags Nested(
    key LowCardinality(String),
    valueString LowCardinality(String),
    valueBool UInt8,
    valueInt Int64,
    valueFloat Float64
  ),
  INDEX tags_strings (tags.key, tags.valueString) TYPE set(0) GRANULARITY 64,
  INDEX tags_ints (tags.key, tags.valueInt) TYPE set(0) GRANULARITY 64
) ENGINE MergeTree() PARTITION BY toDate(timestamp) ORDER BY (timestamp, service, operation);

• Data table for span storage and quick retrieval by traceID

CREATE TABLE jaeger_spans (
  timestamp DateTime64(6),
  traceID FixedString(16),
  model String
) ENGINE MergeTree() PARTITION BY toDate(timestamp) ORDER BY traceID;

You probably need Clickhouse 20.x for DateTime64, I used 20.1.11.73.

Index table looks like this:

SELECT *
FROM jaeger_index
ARRAY JOIN tags
ORDER BY timestamp DESC
LIMIT 20
FORMAT PrettyCompactMonoBlock

┌──────────────────timestamp─┬─traceID──────────┬─service──────┬─operation────┬─durationUs─┬─tags.key─────────────┬─tags.valueString─┬─tags.valueBool─┬─tags.valueInt─┬─tags.valueFloat─┐
│ 2020-05-10 20:43:23.248314 │ 212e5c616f4b9c2f │ jaeger-query │ getTraces    │     131605 │ num_trace_ids        │                  │              0 │            13 │               0 │
│ 2020-05-10 20:43:23.248314 │ 212e5c616f4b9c2f │ jaeger-query │ getTraces    │     131605 │ weird                │                  │              1 │             0 │               0 │
│ 2020-05-10 20:43:23.248314 │ 212e5c616f4b9c2f │ jaeger-query │ getTraces    │     131605 │ π                    │                  │              0 │             0 │            3.14 │
│ 2020-05-10 20:43:23.248314 │ 212e5c616f4b9c2f │ jaeger-query │ getTraces    │     131605 │ internal.span.format │ proto            │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.248314 │ 212e5c616f4b9c2f │ jaeger-query │ getTraces    │     131605 │ jaeger.version       │ Go-2.22.1        │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.248314 │ 212e5c616f4b9c2f │ jaeger-query │ getTraces    │     131605 │ hostname             │ C02TV431HV2Q     │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.248314 │ 212e5c616f4b9c2f │ jaeger-query │ getTraces    │     131605 │ ip                   │ 192.168.1.43     │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.248314 │ 212e5c616f4b9c2f │ jaeger-query │ getTraces    │     131605 │ client-uuid          │ 7fc8f98ddbcd358c │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065577 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraceIDs │     182728 │ internal.span.format │ proto            │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065577 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraceIDs │     182728 │ jaeger.version       │ Go-2.22.1        │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065577 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraceIDs │     182728 │ hostname             │ C02TV431HV2Q     │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065577 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraceIDs │     182728 │ ip                   │ 192.168.1.43     │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065577 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraceIDs │     182728 │ client-uuid          │ 7fc8f98ddbcd358c │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065574 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraces   │     314349 │ internal.span.format │ proto            │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065574 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraces   │     314349 │ jaeger.version       │ Go-2.22.1        │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065574 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraces   │     314349 │ hostname             │ C02TV431HV2Q     │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065574 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraces   │     314349 │ ip                   │ 192.168.1.43     │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065574 │ 212e5c616f4b9c2f │ jaeger-query │ FindTraces   │     314349 │ client-uuid          │ 7fc8f98ddbcd358c │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065535 │ 212e5c616f4b9c2f │ jaeger-query │ /api/traces  │     315554 │ sampler.type         │ const            │              0 │             0 │               0 │
│ 2020-05-10 20:43:23.065535 │ 212e5c616f4b9c2f │ jaeger-query │ /api/traces  │     315554 │ sampler.param        │                  │              1 │             0 │               0 │
└────────────────────────────┴──────────────────┴──────────────┴──────────────┴────────────┴──────────────────────┴──────────────────┴────────────────┴───────────────┴─────────────────┘

Tags are stored in their original types, so with enough SQL-fu you can find all spans with response size between X and Y bytes, for example.

The layout of the query is different from Elasticsearch, since now you have all tags for the trace laid out on a single view. This means that if you search for all operations of some service, you will get a cross-span result where one tag can match one span, and another tag can match another span. Consider the following trace:

+ upstream (tags: {"host": "foo.bar"})
++ upstream_ttfb (tags: {"status": 200})
++ upstream_download (tags: {"error": true})

You can search for host=foo.bar status=200 across all operations and this trace will be found, even though no since span has both tags. This seems like a really nice upside.

There's support for both JSON and Protobuf storage. The former allows out-of-band queries, since Clickhouse supports JSON functions. The latter is much more compact.

I pushed 100K spans from tracegen through this with a local Clickhouse in a Docker container with stock settings, and here's how storage looks like:

SELECT
    table,
    sum(marks) AS marks,
    sum(rows) AS rows,
    sum(bytes_on_disk) AS bytes_on_disk,
    sum(data_compressed_bytes) AS data_compressed_bytes,
    sum(data_uncompressed_bytes) AS data_uncompressed_bytes,
    toDecimal64(data_uncompressed_bytes / data_compressed_bytes, 2) AS compression_ratio,
    toDecimal64(data_compressed_bytes / rows, 2) AS compressed_bytes_per_row
FROM system.parts
WHERE table LIKE 'jaeger_%'
GROUP BY table
ORDER BY table ASC

SELECT
    table,
    sum(marks) AS marks,
    sum(rows) AS rows,
    sum(bytes_on_disk) AS bytes_on_disk,
    sum(data_compressed_bytes) AS data_compressed_bytes,
    sum(data_uncompressed_bytes) AS data_uncompressed_bytes,
    toDecimal64(data_uncompressed_bytes / data_compressed_bytes, 2) AS compression_ratio,
    toDecimal64(data_compressed_bytes / rows, 2) AS compressed_bytes_per_row
FROM system.parts
WHERE table LIKE 'jaeger_%'
GROUP BY table
ORDER BY table ASC

┌─table────────┬─marks─┬───rows─┬─bytes_on_disk─┬─data_compressed_bytes─┬─data_uncompressed_bytes─┬─compression_ratio─┬─compressed_bytes_per_row─┐
│ jaeger_index │    16 │ 106667 │       2121539 │               2110986 │                22678493 │             10.74 │                    19.79 │
│ jaeger_spans │    20 │ 106667 │       5634663 │               5632817 │                37112272 │              6.58 │                    52.80 │
└──────────────┴───────┴────────┴───────────────┴───────────────────────┴─────────────────────────┴───────────────────┴──────────────────────────┘

SELECT
    table,
    column,
    type,
    sum(column_data_compressed_bytes) AS compressed,
    sum(column_data_uncompressed_bytes) AS uncompressed,
    toDecimal64(uncompressed / compressed, 2) AS compression_ratio,
    sum(rows) AS rows,
    toDecimal64(compressed / rows, 2) AS bytes_per_row
FROM system.parts_columns
WHERE (table LIKE 'jaeger_%') AND active
GROUP BY
    table,
    column,
    type
ORDER BY
    table ASC,
    column ASC

┌─table────────┬─column───────────┬─type──────────────────────────┬─compressed─┬─uncompressed─┬─compression_ratio─┬───rows─┬─bytes_per_row─┐
│ jaeger_index │ durationUs       │ UInt64                        │     248303 │       853336 │              3.43 │ 106667 │          2.32 │
│ jaeger_index │ operation        │ LowCardinality(String)        │       5893 │       107267 │             18.20 │ 106667 │          0.05 │
│ jaeger_index │ service          │ LowCardinality(String)        │        977 │       107086 │            109.60 │ 106667 │          0.00 │
│ jaeger_index │ tags.key         │ Array(LowCardinality(String)) │      29727 │      1811980 │             60.95 │ 106667 │          0.27 │
│ jaeger_index │ tags.valueBool   │ Array(UInt8)                  │      29063 │      1810904 │             62.30 │ 106667 │          0.27 │
│ jaeger_index │ tags.valueFloat  │ Array(Float64)                │      44762 │      8513880 │            190.20 │ 106667 │          0.41 │
│ jaeger_index │ tags.valueInt    │ Array(Int64)                  │     284393 │      8513880 │             29.93 │ 106667 │          2.66 │
│ jaeger_index │ tags.valueString │ Array(LowCardinality(String)) │      31695 │      1814416 │             57.24 │ 106667 │          0.29 │
│ jaeger_index │ timestamp        │ DateTime64(6)                 │     431835 │       853336 │              1.97 │ 106667 │          4.04 │
│ jaeger_index │ traceID          │ FixedString(16)               │    1063375 │      1706672 │              1.60 │ 106667 │          9.96 │
│ jaeger_spans │ model            │ String                        │    4264180 │     34552264 │              8.10 │ 106667 │         39.97 │
│ jaeger_spans │ timestamp        │ DateTime64(6)                 │     463444 │       853336 │              1.84 │ 106667 │          4.34 │
│ jaeger_spans │ traceID          │ FixedString(16)               │     905193 │      1706672 │              1.88 │ 106667 │          8.48 │
└──────────────┴──────────────────┴───────────────────────────────┴────────────┴──────────────┴───────────────────┴────────┴───────────────┘

We have around 74B daily docs in our production Elasticsearch storage. My plan is to switch that to fields-as-tags, remove indexing of non-queried fields (logs, nested tags, references), then switch to a sorted index and then see how Clickhouse compares to that for the same spans.

[yurishkuro]

@bobrik very interesting, thanks for sharing. I am curious what the performance for retrieving by trace ID would be like.

Q: why do you use LowCardinality(String) for tags? Some tags can be very high cardinality, e.g. URLs.

You can search for host=foo.bar status=200 across all operations and this trace will be found, even though no since span has both tags. This seems like a really nice upside.

I'm confused why this would be the case. Doesn't CH evaluate the query in full against each row (i.e. each span)? Or is this because how your plugin interacts with CH?

[Slach]

@yurishkuro even 100k per block cardinality, LowCardinality(String) with dictionary based encoding will better than just String