ingest consumer: more granular error handling, committer sanity check #6951
Conversation
* check that the offset we're committing is certainly from the partition we're committing to * process a fetch even when it contains some errors; this allows to process fetches with partial data Signed-off-by: Dimitar Dimitrov <dimitar.dimitrov@grafana.com>
| r.recordFetchesMetrics(fetches) | ||
| r.logFetchErrs(fetches) |
There was a problem hiding this comment.
I'm still not super convinced about this approach. I think mixing error and non-error fetches in these functions could be error prone. What if we construct a slice of successful fetches (which is the slice returned by PollFetches() if there are no errors) and then we call recordFetchesMetrics(), consumeFetches() and enqueueCommit() passing only the successful fetches?
There was a problem hiding this comment.
the fetches abstraction is difficult to work with in pure go in the first place and we use its iterators everywhere (iterating partitions or records), but maybe filtering out the error fetches makes reasoning later on easier; I did that in the latest commit, PTAL
| @@ -341,6 +340,10 @@ func newReaderMetrics(partitionID int32, reg prometheus.Registerer) readerMetric | |||
| Help: "The number of records received by the consumer in a single fetch operation.", | |||
| Buckets: prometheus.ExponentialBuckets(1, 2, 15), | |||
| }), | |||
| fetchesErrors: factory.NewCounter(prometheus.CounterOpts{ | |||
| Name: "cortex_ingest_storage_reader_fetch_errors_total", | |||
There was a problem hiding this comment.
Do we already have a metric with the total number of fetches, to compute a % of failing ones?
There was a problem hiding this comment.
maybe we can infer something from the franz-go metrics, but i prefer to avoid that (is a batch the same as a fetch?). I added a counter for the number of fetches too.
Signed-off-by: Dimitar Dimitrov <dimitar.dimitrov@grafana.com>
Signed-off-by: Dimitar Dimitrov <dimitar.dimitrov@grafana.com>
Signed-off-by: Dimitar Dimitrov <dimitar.dimitrov@grafana.com>
| level.Error(r.logger).Log("msg", "encountered error while fetching", "err", err) | ||
| continue | ||
| } | ||
|
|
||
| r.recordFetchesMetrics(fetches) |
There was a problem hiding this comment.
Could some of the metrics falsed if we track them for error-ed fetches too?
There was a problem hiding this comment.
I'm thinking to receiveDelay in particular.
There was a problem hiding this comment.
whenever the fetch has an error it never has records; we can perhaps have misleading values for the number of records per fetch and the number of fetches we do from kafka. Since some error fetches can be synthetic (i.e. not actually returned from kafka), i moved metrics recording after filtering out error fetches
it's difficult with |
dimitarvdimitrov
deleted the
dimitar/ingest/address-post-merge-fb-on-6929
branch
Follow-up of #6929