Reduce duplication when writing

[bboreham]

We replicate writes to N (typically 3) ingesters to reduce data loss from random failures of machine or process. However this means in stable operation each metric is written to store three times, along with index entries for each label (at least in the DynamoDB back-end). Say 10 index entries per chunk, for a total of 33 writes. Nearly one third of these writes are pointless and, worse, they slow down queries because all those entries get re-fetched and finally deduplicated.

One alternative idea:
When an ingester is about to write a chunk, query the index to see if an overlapping chunk already exists. If so, subtract the overlapping part and merge the remainder back into the chunks being accumulated. This would replace the 33 writes with 11 writes and 2 reads.

This idea doesn't require any particular consistency from the store - if it under-reports we'll just store some pointless chunks as before.

[jml]

The details have gone hazy, but I believe there's a trade-off with how many ingesters we need to read from. The Dynamo paper section 4.5 has some discussion on this.

[tomwilkie]

This proposal makes every write into a read + a write, and potentially a delete. As Cortex tends to do many orders-of-magnitude more writes than reads, making queries cheaper by making writes more expensive seems like the wrong direction...

It might be worth considering TCO here; for systems like BigTable & Cassandra, writes are significantly cheaper than reads (in turns of IOPS and $$$), so doing read-on-write is a big no. Its interesting that this isn't the case for DynamoDB - reads seem to be 1/5th the cost - but does at least suggest that this needs to be done on a per-backend basis. Although the existing abstractions won't nicely support that.

Alternatives to consider:

• The 3x index cost should probably be tackled by indirecting via a timeseries ID, and not having the entries point straight at chunks. This is something we started with @aaron7 in Use LazySeriesIterator with fuzzy metric name queries #442, using the sha256 of the label-set of a series as its fingerprint. This might also be necessary to do proper stream processing of queries anyway, but won't cover the 3x chunk cost.
• On a query you could always get the range of chunks and then read at most 2 of the 3 chunks for every overlapping range, but this seems like it might be challenging (fun?) to compute.
• We don't actually need to write to 3 ingesters in normal operation, we should only write to two. This is discussed in Don't try and read/write all ingesters in a set; only do it if one fails/times out #100, would get you 50% of the wins, and is probably relatively easy to implement.
• We could potentially do the merge as form of "Read Repair", writing back deduped chunks after a query. This would work for backends like Cassandra and BigTable, as you've already paid the read cost.
• Moving to something like TSDB for the ingesters, and guaranteeing the last N hours are in an ingester, would make >80% of queries not touch the chunk store. I think users are willing to accept slower queries for older data.

As a less rational fear, getting the write path to perform reliably was really challenging; I'd be pretty apprehensive to adding a read into that pipeline.

[bboreham]

Thanks, I appreciate the insights.

Bit more background: the main use-case I'm targeting is when you see some data in the last hour and think "what does that look like over the last month?". Depending on the metric, the index reads for 30 days can be very expensive.

I would not change the write path lightly.

[tomwilkie]

Interesting - I wouldn't have thought hour-long queries in the past would have been any more expensive than hour-long now queries, modulo caching effects. Are the historical tables adequately provisioned in DynamoDB? Do you know if its index lookups or chunk fetches that are dominating?

If its index, I remember an optimisation I added to make index lookups from [now-1hr, now) cheaper - the index always looks up to the end of the "row". This is indicated by the inclusion of only a greater-than comparison here. Given we know the maximum size of chunks, it would be possible to add a less-than comparison to the same place (and plumbing through the logic for the schemas), and terminate the scan early. Would be a minor improvement for queries that happen in the morning...

If its chunk fetches taking the majority of time (which I guess it is by your recent PRs), IIRC is the shear number of chunks that need fetching, more than the size of the data, that causes problems. Do you see a lot of DynamoDB throttling? You could consider:

• proactively prefetching month-ago historic chunks on queries in the background
• When I get round to doing the TSDB work the average chunk size will increase significantly (and should always hit max chunk age), which should massively reduce the number of chunks fetched.
• Parallelising chunk fetches across multiple parallel workers, and have them stream deduped series back to the querier.

Alternatively, instead of putting this kind of deduping and processing on the write path, you could instead do it as a separate process which reads back data and rewrites it asynchronously, taking it off the critical path. The original design doc included a mention of this IIRC.

[bboreham]

Index queries can be several seconds each. The data for one bucket for one index all comes from one hash key, so I'm not sure more provisioning helps - just fetching 100,000 data items takes time. Even filtering them takes 1-2 seconds.

In many cases that number can be reduced with better query formation; I have added some thoughts to #209. This was another thought that came to me.

I am working on many fronts; the PRs you see are just the parts that are ready now.

[bboreham]

To clarify, "what does that look like over the last month" is a 30-day query not an hour-long query.

But an hour-long query in the past is also slower, because we read to the end of the bucket.

[bboreham]

Since we're on the subject, #209 says " don't range query to end of row, but use max chunk length to query to end of query + 12hrs"; what is that "+ 12hrs" for?

[tomwilkie]

Index queries can be several seconds each. The data for one bucket for one index all comes from one hash key, so I'm not sure more provisioning helps - just fetching 100,000 data items takes time. Even filtering them takes 1-2 seconds.

Indeed; indirection through a "time series" index would massively improve that, as the labels index would only contain the cardinality of the labels, not cardinality x number of chunks.

In many cases that number can be reduced with better query formation; I have added some thoughts to #209. This was another thought that came to me.

Yes! Even better, you could track and pick low cardinality indexes and only read them; you can always under read the index and over read chunks, and filter later. Thats if the index is slower than the chunks of course.

To clarify, "what does that look like over the last month" is a 30-day query not an hour-long query.

Ah I see. How slow?

Another idea is to pipeline index and chunk fetches, so we can start fetching chunks earlier. Thats a fair bit of work mind you.

Since we're on the subject, #209 says " don't range query to end of row, but use max chunk length to query to end of query + 12hrs"; what is that "+ 12hrs" for?

12hrs is max chunk age

[tomwilkie]

#875 now writes out (label:value) -> []seriesID and seriesID -> []chunkID indexes, which reduces the size of the index by 6-12x over the (label:value) -> []chunkID indexes of old - 3x due to replication and 4x due to average chunk size of 6hrs vs 24hrs index window.

We don't dedupe chunks yet, but this should massively reduce size of index.

[tomwilkie]

Given #875 is merged, can this be closed?

[bboreham]

The original point hasn't changed - if the chunk you're about to write is already in the store then you would benefit from not re-writing it.

Querying the disk store wasn't a good idea, but querying memcache should work better.

You have very different usage patterns if you see an "average chunk size of 6hrs". 2 hours is more like what we see as an overall average, but if you are scraping a cluster where containers start and stop a lot then the average goes down. Chunks with just a few minutes of data are ideal candidates for de-duping, because they will have identical start and end times on all replicas.