mirror of
https://github.com/prometheus/prometheus.git
synced 2024-11-09 23:24:05 -08:00
aa3513fc89
* consoles: exclude iowait and steal from CPU Utilisation 'iowait' and 'steal' indicate specific idle/wait states, which shouldn't be counted into CPU Utilisation. Also see https://github.com/prometheus-operator/kube-prometheus/pull/796 and https://github.com/kubernetes-monitoring/kubernetes-mixin/pull/667. Per the iostat man page: %idle Show the percentage of time that the CPU or CPUs were idle and the system did not have an outstanding disk I/O request. %iowait Show the percentage of time that the CPU or CPUs were idle during which the system had an outstanding disk I/O request. %steal Show the percentage of time spent in involuntary wait by the virtual CPU or CPUs while the hypervisor was servicing another virtual processor. Signed-off-by: Julian Wiedmann <jwi@linux.ibm.com> * tsdb: shrink txRing with smaller integers 4 billion active transactions ought to be enough for anyone. Signed-off-by: Bryan Boreham <bjboreham@gmail.com> * tsdb: create isolation transaction slice on demand When Prometheus restarts it creates every series read in from the WAL, but many of those series will be finished, and never receive any more samples. By defering allocation of the txRing slice to when it is first needed, we save 32 bytes per stale series. Signed-off-by: Bryan Boreham <bjboreham@gmail.com> * add cluster variable to Overview dashboard Signed-off-by: Erik Sommer <ersotech@posteo.de> * promql: simplify Native Histogram arithmetics Signed-off-by: Linas Medziunas <linas.medziunas@gmail.com> * Cut 2.49.0-rc.0 (#13270) * Cut 2.49.0-rc.0 Signed-off-by: bwplotka <bwplotka@gmail.com> * Removed the duplicate. Signed-off-by: bwplotka <bwplotka@gmail.com> --------- Signed-off-by: bwplotka <bwplotka@gmail.com> * Add unit protobuf parser Signed-off-by: Arianna Vespri <arianna.vespri@yahoo.it> * Go on adding protobuf parsing for unit Signed-off-by: Arianna Vespri <arianna.vespri@yahoo.it> * ui: create a reproduction for https://github.com/prometheus/prometheus/issues/13292 Signed-off-by: machine424 <ayoubmrini424@gmail.com> * Get conditional right Signed-off-by: Arianna Vespri <arianna.vespri@yahoo.it> * Get VM Scale Set NIC (#13283) Calling `*armnetwork.InterfacesClient.Get()` doesn't work for Scale Set VM NIC, because these use a different Resource ID format. Use `*armnetwork.InterfacesClient.GetVirtualMachineScaleSetNetworkInterface()` instead. This needs both the scale set name and the instance ID, so add an `InstanceID` field to the `virtualMachine` struct. `InstanceID` is empty for a VM that isn't a ScaleSetVM. Signed-off-by: Daniel Nicholls <daniel.nicholls@resdiary.com> * Cut v2.49.0-rc.1 Signed-off-by: bwplotka <bwplotka@gmail.com> * Delete debugging lines, amend error message for unit Signed-off-by: Arianna Vespri <arianna.vespri@yahoo.it> * Correct order in error message Signed-off-by: Arianna Vespri <arianna.vespri@yahoo.it> * Consider storage.ErrTooOldSample as non-retryable Signed-off-by: Daniel Kerbel <nmdanny@gmail.com> * scrape_test.go: Increase scrape interval in TestScrapeLoopCache to reduce potential flakiness Signed-off-by: machine424 <ayoubmrini424@gmail.com> * Avoid creating string for suffix, consider counters without _total suffix Signed-off-by: Arianna Vespri <arianna.vespri@yahoo.it> * build(deps): bump github.com/prometheus/client_golang Bumps [github.com/prometheus/client_golang](https://github.com/prometheus/client_golang) from 1.17.0 to 1.18.0. - [Release notes](https://github.com/prometheus/client_golang/releases) - [Changelog](https://github.com/prometheus/client_golang/blob/main/CHANGELOG.md) - [Commits](https://github.com/prometheus/client_golang/compare/v1.17.0...v1.18.0) --- updated-dependencies: - dependency-name: github.com/prometheus/client_golang dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * build(deps): bump actions/setup-node from 3.8.1 to 4.0.1 Bumps [actions/setup-node](https://github.com/actions/setup-node) from 3.8.1 to 4.0.1. - [Release notes](https://github.com/actions/setup-node/releases) - [Commits](5e21ff4d9b...b39b52d121
) --- updated-dependencies: - dependency-name: actions/setup-node dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> * scripts: sort file list in embed directive Otherwise the resulting string depends on find, which afaict depends on the underlying filesystem. A stable file list make it easier to detect UI changes in downstreams that need to track UI assets. Signed-off-by: Jan Fajerski <jfajersk@redhat.com> * Fix DataTableProps['data'] for resultType string Signed-off-by: Kevin Mingtarja <kevin.mingtarja@gmail.com> * Fix handling of scalar and string in isHeatmapData Signed-off-by: Kevin Mingtarja <kevin.mingtarja@gmail.com> * build(deps): bump github.com/influxdata/influxdb Bumps [github.com/influxdata/influxdb](https://github.com/influxdata/influxdb) from 1.11.2 to 1.11.4. - [Release notes](https://github.com/influxdata/influxdb/releases) - [Commits](https://github.com/influxdata/influxdb/compare/v1.11.2...v1.11.4) --- updated-dependencies: - dependency-name: github.com/influxdata/influxdb dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> * build(deps): bump github.com/prometheus/prometheus Bumps [github.com/prometheus/prometheus](https://github.com/prometheus/prometheus) from 0.48.0 to 0.48.1. - [Release notes](https://github.com/prometheus/prometheus/releases) - [Changelog](https://github.com/prometheus/prometheus/blob/main/CHANGELOG.md) - [Commits](https://github.com/prometheus/prometheus/compare/v0.48.0...v0.48.1) --- updated-dependencies: - dependency-name: github.com/prometheus/prometheus dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> * Bump client_golang to v1.18.0 (#13373) Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Drop old inmemory samples (#13002) * Drop old inmemory samples Co-authored-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Avoid copying timeseries when the feature is disabled Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Run gofmt Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Clarify docs Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Add more logging info Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Remove loggers Signed-off-by: Marc Tuduri <marctc@protonmail.com> * optimize function and add tests Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Simplify filter Signed-off-by: Marc Tuduri <marctc@protonmail.com> * rename var Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Update help info from metrics Signed-off-by: Marc Tuduri <marctc@protonmail.com> * use metrics to keep track of drop elements during buildWriteRequest Signed-off-by: Marc Tuduri <marctc@protonmail.com> * rename var in tests Signed-off-by: Marc Tuduri <marctc@protonmail.com> * pass time.Now as parameter Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Change buildwriterequest during retries Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Revert "Remove loggers" This reverts commit 54f91dfcae20488944162335ab4ad8be459df1ab. Signed-off-by: Marc Tuduri <marctc@protonmail.com> * use log level debug for loggers Signed-off-by: Marc Tuduri <marctc@protonmail.com> * Fix linter Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Remove noisy debug-level logs; add 'reason' label to drop metrics Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Remove accidentally committed files Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Propagate logger to buildWriteRequest to log dropped data Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Fix docs comment Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Make drop reason more specific Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Remove unnecessary pass of logger Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Use snake_case for reason label Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> * Fix dropped samples metric Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> --------- Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> Signed-off-by: Marc Tuduri <marctc@protonmail.com> Signed-off-by: Paschalis Tsilias <tpaschalis@users.noreply.github.com> Co-authored-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> Co-authored-by: Paschalis Tsilias <tpaschalis@users.noreply.github.com> * fix(discovery): allow requireUpdate util to timeout in discovery/file/file_test.go. The loop ran indefinitely if the condition isn't met. Before, each iteration created a new timer channel which was always outpaced by the other timer channel with smaller duration. minor detail: There was a memory leak: resources of the ~10 previous timers were constantly kept. With the fix, we may keep the resources of one timer around for defaultWait but this isn't worth the changes to make it right. Signed-off-by: machine424 <ayoubmrini424@gmail.com> * Merge pull request #13371 from kevinmingtarja/fix-isHeatmapData ui: fix handling of scalar and string in isHeatmapData * tsdb/{index,compact}: allow using custom postings encoding format (#13242) * tsdb/{index,compact}: allow using custom postings encoding format We would like to experiment with a different postings encoding format in Thanos so in this change I am proposing adding another argument to `NewWriter` which would allow users to change the format if needed. Also, wire the leveled compactor so that it would be possible to change the format there too. Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * tsdb/compact: use a struct for leveled compactor options As discussed on Slack, let's use a struct for the options in leveled compactor. Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * tsdb: make changes after Bryan's review - Make changes less intrusive - Turn the postings encoder type into a function - Add NewWriterWithEncoder() Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> --------- Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * Cut 2.49.0-rc.2 Signed-off-by: bwplotka <bwplotka@gmail.com> * build(deps): bump actions/setup-go from 3.5.0 to 5.0.0 in /scripts (#13362) Bumps [actions/setup-go](https://github.com/actions/setup-go) from 3.5.0 to 5.0.0. - [Release notes](https://github.com/actions/setup-go/releases) - [Commits](6edd4406fa...0c52d547c9
) --- updated-dependencies: - dependency-name: actions/setup-go dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * build(deps): bump github/codeql-action from 2.22.8 to 3.22.12 (#13358) Bumps [github/codeql-action](https://github.com/github/codeql-action) from 2.22.8 to 3.22.12. - [Release notes](https://github.com/github/codeql-action/releases) - [Changelog](https://github.com/github/codeql-action/blob/main/CHANGELOG.md) - [Commits](407ffafae6...012739e508
) --- updated-dependencies: - dependency-name: github/codeql-action dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * put @nexucis has a release shepherd (#13383) Signed-off-by: Augustin Husson <augustin.husson@amadeus.com> * Add analyze histograms command to promtool (#12331) Add `query analyze` command to promtool This command analyzes the buckets of classic and native histograms, based on data queried from the Prometheus query API, i.e. it doesn't require direct access to the TSDB files. Signed-off-by: Jeanette Tan <jeanette.tan@grafana.com> --------- Signed-off-by: Jeanette Tan <jeanette.tan@grafana.com> * included instance in all necessary descriptions Signed-off-by: Erik Sommer <ersotech@posteo.de> * tsdb/compact: fix passing merge func Fixing a very small logical problem I've introduced :(. Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * tsdb: add enable overlapping compaction This functionality is needed in downstream projects because they have a separate component that does compaction. Upstreaming7c8e9a2a76/tsdb/compact.go (L323-L325)
. Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * Cut 2.49.0 Signed-off-by: bwplotka <bwplotka@gmail.com> * promtool: allow setting multiple matchers to "promtool tsdb dump" command. (#13296) Conditions are ANDed inside the same matcher but matchers are ORed Including unit tests for "promtool tsdb dump". Refactor some matchers scraping utils. Signed-off-by: machine424 <ayoubmrini424@gmail.com> * Fixed changelog Signed-off-by: bwplotka <bwplotka@gmail.com> * tsdb/main: wire "EnableOverlappingCompaction" to tsdb.Options (#13398) This added the https://github.com/prometheus/prometheus/pull/13393 "EnableOverlappingCompaction" parameter to the compactor code but not to the tsdb.Options. I forgot about that. Add it to `tsdb.Options` too and set it to `true` in Prometheus. Copy/paste the description from https://github.com/prometheus/prometheus/pull/13393#issuecomment-1891787986 Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * Issue #13268: fix quality value in accept header Signed-off-by: Kumar Kalpadiptya Roy <kalpadiptya.roy@outlook.com> * Cut 2.49.1 with scrape q= bugfix. Signed-off-by: bwplotka <bwplotka@gmail.com> * Cut 2.49.1 web package. Signed-off-by: bwplotka <bwplotka@gmail.com> * Restore more efficient version of NewPossibleNonCounterInfo annotation (#13022) Restore more efficient version of NewPossibleNonCounterInfo annotation Signed-off-by: Jeanette Tan <jeanette.tan@grafana.com> --------- Signed-off-by: Jeanette Tan <jeanette.tan@grafana.com> * Fix regressions introduced by #13242 Signed-off-by: Marco Pracucci <marco@pracucci.com> * fix slice copy in 1.20 (#13389) The slices package is added to the standard library in Go 1.21; we need to import from the exp area to maintain compatibility with Go 1.20. Signed-off-by: tyltr <tylitianrui@126.com> * Docs: Query Basics: link to rate (#10538) Co-authored-by: Julien Pivotto <roidelapluie@o11y.eu> * chore(kubernetes): check preconditions earlier and avoid unnecessary checks or iterations Signed-off-by: machine424 <ayoubmrini424@gmail.com> * Examples: link to `rate` for new users (#10535) * Examples: link to `rate` for new users Signed-off-by: Ted Robertson 10043369+tredondo@users.noreply.github.com Co-authored-by: Bryan Boreham <bjboreham@gmail.com> * promql: use natural sort in sort_by_label and sort_by_label_desc (#13411) These functions are intended for humans, as robots can already sort the results however they please. Humans like things sorted "naturally": * https://blog.codinghorror.com/sorting-for-humans-natural-sort-order/ A similar thing has been done to Grafana, which is also used by humans: * https://github.com/grafana/grafana/pull/78024 * https://github.com/grafana/grafana/pull/78494 Signed-off-by: Ivan Babrou <github@ivan.computer> * TestLabelValuesWithMatchers: Add test case Signed-off-by: Arve Knudsen <arve.knudsen@gmail.com> * remove obsolete build tag Signed-off-by: tyltr <tylitianrui@126.com> * Upgrade some golang dependencies for resty 2.11 Signed-off-by: Israel Blancas <iblancasa@gmail.com> * Native Histograms: support `native_histogram_min_bucket_factor` in scrape_config (#13222) Native Histograms: support native_histogram_min_bucket_factor in scrape_config --------- Signed-off-by: Ziqi Zhao <zhaoziqi9146@gmail.com> Signed-off-by: Björn Rabenstein <github@rabenste.in> Co-authored-by: George Krajcsovits <krajorama@users.noreply.github.com> Co-authored-by: Björn Rabenstein <github@rabenste.in> * Add warnings for histogramRate applied with isCounter not matching counter/gauge histogram (#13392) Add warnings for histogramRate applied with isCounter not matching counter/gauge histogram --------- Signed-off-by: Jeanette Tan <jeanette.tan@grafana.com> * Minor fixes to otlp vendor update script Signed-off-by: Goutham <gouthamve@gmail.com> * build(deps): bump github.com/hetznercloud/hcloud-go/v2 Bumps [github.com/hetznercloud/hcloud-go/v2](https://github.com/hetznercloud/hcloud-go) from 2.4.0 to 2.6.0. - [Release notes](https://github.com/hetznercloud/hcloud-go/releases) - [Changelog](https://github.com/hetznercloud/hcloud-go/blob/main/CHANGELOG.md) - [Commits](https://github.com/hetznercloud/hcloud-go/compare/v2.4.0...v2.6.0) --- updated-dependencies: - dependency-name: github.com/hetznercloud/hcloud-go/v2 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * Enhanced visibility for `promtool test rules` with JSON colored formatting (#13342) * Added diff flag for unit test to improvise readability & debugging Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * Removed blank spaces Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * Fixed linting error Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * Added cli flags to documentation Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * Revert unrrelated linting fixes Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * Fixed review suggestions Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * Cleanup Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * Updated flag description Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * Updated flag description Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> --------- Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> * storage: skip merging when no remote storage configured Prometheus is hard-coded to use a fanout storage between TSDB and a remote storage which by default is empty. This change detects the empty storage and skips merging between result sets, which would make `Select()` sort results. Bottom line: we skip a sort unless there really is some remote storage configured. Signed-off-by: Bryan Boreham <bjboreham@gmail.com> * Remove csmarchbanks from remote write owners (#13432) I have not had the time to keep up with remote write and have no plans to work on it in the near future so I am withdrawing my maintainership of that part of the codebase. I continue to focus on client_python. Signed-off-by: Chris Marchbanks <csmarchbanks@gmail.com> * add more context cancellation check at evaluation time Signed-off-by: Ben Ye <benye@amazon.com> * Optimize label values with matchers by taking shortcuts (#13426) Don't calculate postings beforehand: we may not need them. If all matchers are for the requested label, we can just filter its values. Also, if there are no values at all, no need to run any kind of logic. Also add more labelValuesWithMatchers benchmarks Signed-off-by: Oleg Zaytsev <mail@olegzaytsev.com> * Add automatic memory limit handling Enable automatic detection of memory limits and configure GOMEMLIMIT to match. * Also includes a flag to allow controlling the reserved ratio. Signed-off-by: SuperQ <superq@gmail.com> * Update OSSF badge link (#13433) Provide a more user friendly interface Signed-off-by: Matthieu MOREL <matthieu.morel35@gmail.com> * SD Managers taking over responsibility for registration of debug metrics (#13375) SD Managers take over responsibility for SD metrics registration --------- Signed-off-by: Paulin Todev <paulin.todev@gmail.com> Signed-off-by: Björn Rabenstein <github@rabenste.in> Co-authored-by: Björn Rabenstein <github@rabenste.in> * Optimize histogram iterators (#13340) Optimize histogram iterators Histogram iterators allocate new objects in the AtHistogram and AtFloatHistogram methods, which makes calculating rates over long ranges expensive. In #13215 we allowed an existing object to be reused when converting an integer histogram to a float histogram. This commit follows the same idea and allows injecting an existing object in the AtHistogram and AtFloatHistogram methods. When the injected value is nil, iterators allocate new histograms, otherwise they populate and return the injected object. The commit also adds a CopyTo method to Histogram and FloatHistogram which is used in the BufferedIterator to overwrite items in the ring instead of making new copies. Note that a specialized HPoint pool is needed for all of this to work (`matrixSelectorHPool`). --------- Signed-off-by: Filip Petkovski <filip.petkovsky@gmail.com> Co-authored-by: George Krajcsovits <krajorama@users.noreply.github.com> * doc: Mark `mad_over_time` as experimental (#13440) We forgot to do that in https://github.com/prometheus/prometheus/pull/13059 Signed-off-by: beorn7 <beorn@grafana.com> * Change metric label for Puppetdb from 'http' to 'puppetdb' Signed-off-by: Paulin Todev <paulin.todev@gmail.com> * mirror metrics.proto change & generate code Signed-off-by: Ziqi Zhao <zhaoziqi9146@gmail.com> * TestHeadLabelValuesWithMatchers: Add test case (#13414) Add test case to TestHeadLabelValuesWithMatchers, while fixing a couple of typos in other test cases. Also enclosing some implicit sub-tests in a `t.Run` call to make them explicitly sub-tests. Signed-off-by: Arve Knudsen <arve.knudsen@gmail.com> * update all go dependencies (#13438) Signed-off-by: Augustin Husson <husson.augustin@gmail.com> * build(deps): bump the k8s-io group with 2 updates (#13454) Bumps the k8s-io group with 2 updates: [k8s.io/api](https://github.com/kubernetes/api) and [k8s.io/client-go](https://github.com/kubernetes/client-go). Updates `k8s.io/api` from 0.28.4 to 0.29.1 - [Commits](https://github.com/kubernetes/api/compare/v0.28.4...v0.29.1) Updates `k8s.io/client-go` from 0.28.4 to 0.29.1 - [Changelog](https://github.com/kubernetes/client-go/blob/master/CHANGELOG.md) - [Commits](https://github.com/kubernetes/client-go/compare/v0.28.4...v0.29.1) --- updated-dependencies: - dependency-name: k8s.io/api dependency-type: direct:production update-type: version-update:semver-minor dependency-group: k8s-io - dependency-name: k8s.io/client-go dependency-type: direct:production update-type: version-update:semver-minor dependency-group: k8s-io ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * build(deps): bump the go-opentelemetry-io group with 1 update (#13453) Bumps the go-opentelemetry-io group with 1 update: [go.opentelemetry.io/collector/semconv](https://github.com/open-telemetry/opentelemetry-collector). Updates `go.opentelemetry.io/collector/semconv` from 0.92.0 to 0.93.0 - [Release notes](https://github.com/open-telemetry/opentelemetry-collector/releases) - [Changelog](https://github.com/open-telemetry/opentelemetry-collector/blob/main/CHANGELOG-API.md) - [Commits](https://github.com/open-telemetry/opentelemetry-collector/compare/v0.92.0...v0.93.0) --- updated-dependencies: - dependency-name: go.opentelemetry.io/collector/semconv dependency-type: direct:production update-type: version-update:semver-minor dependency-group: go-opentelemetry-io ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * build(deps): bump actions/upload-artifact from 3.1.3 to 4.0.0 (#13355) Bumps [actions/upload-artifact](https://github.com/actions/upload-artifact) from 3.1.3 to 4.0.0. - [Release notes](https://github.com/actions/upload-artifact/releases) - [Commits](a8a3f3ad30...c7d193f32e
) --- updated-dependencies: - dependency-name: actions/upload-artifact dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * build(deps): bump bufbuild/buf-push-action (#13357) Bumps [bufbuild/buf-push-action](https://github.com/bufbuild/buf-push-action) from 342fc4cdcf29115a01cf12a2c6dd6aac68dc51e1 to a654ff18effe4641ebea4a4ce242c49800728459. - [Release notes](https://github.com/bufbuild/buf-push-action/releases) - [Commits](342fc4cdcf...a654ff18ef
) --- updated-dependencies: - dependency-name: bufbuild/buf-push-action dependency-type: direct:production ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Labels: Add DropMetricName function, used in PromQL (#13446) This function is called very frequently when executing PromQL functions, and we can do it much more efficiently inside Labels. In the common case that `__name__` comes first in the labels, we simply re-point to start at the next label, which is nearly free. `DropMetricName` is now so cheap I removed the cache - benchmarks show everything still goes faster. Signed-off-by: Bryan Boreham <bjboreham@gmail.com> * tsdb: simplify internal series delete function (#13261) Lifting an optimisation from Agent code, `seriesHashmap.del` can use the unique series reference, doesn't need to check Labels. Also streamline the logic for deleting from `unique` and `conflicts` maps, and add some comments to help the next person. Signed-off-by: Bryan Boreham <bjboreham@gmail.com> * otlptranslator/update-copy.sh: Fix sed command lines Signed-off-by: Arve Knudsen <arve.knudsen@gmail.com> * Rollback k8s.io requirements (#13462) Rollback k8s.io Go modules to v0.28.6 to avoid forcing upgrade of Go to 1.21. This allows us to keep compatibility with the currently supported upstream Go releases. Signed-off-by: SuperQ <superq@gmail.com> * Make update-copy.sh work for both OSX and GNU sed Signed-off-by: Arve Knudsen <arve.knudsen@gmail.com> * Name @beorn7 and @krajorama as maintainers for native histograms I have been the de-facto maintainer for native histograms from the beginning. So let's put this into MAINTAINERS.md. In addition, I hereby proposose George Krajcsovits AKA Krajo as a co-maintainer. He has contributed a lot of native histogram code, but more importantly, he has contributed substantially to reviewing other contributors' native histogram code, up to a point where I was merely rubberstamping the PRs he had already reviewed. I'm confident that he is ready to to be granted commit rights as outlined in the "Maintainers" section of the governance: https://prometheus.io/governance/#maintainers According to the same section of the governance, I will announce the proposed change on the developers mailing list and will give some time for lazy consensus before merging this PR. Signed-off-by: beorn7 <beorn@grafana.com> * ui/fix: correct url handling for stacked graphs (#13460) Signed-off-by: Yury Moladau <yurymolodov@gmail.com> * tsdb: use cheaper Mutex on series Mutex is 8 bytes; RWMutex is 24 bytes and much more complicated. Since `RLock` is only used in two places, `UpdateMetadata` and `Delete`, neither of which are hotspots, we should use the cheaper one. Signed-off-by: Bryan Boreham <bjboreham@gmail.com> * Fix last_over_time for native histograms The last_over_time retains a histogram sample without making a copy. This sample is now coming from the buffered iterator used for windowing functions, and can be reused for reading subsequent samples as the iterator progresses. I would propose copying the sample in the last_over_time function, similar to how it is done for rate, sum_over_time and others. Signed-off-by: Filip Petkovski <filip.petkovsky@gmail.com> * Implementation NOTE: Rebased from main after refactor in #13014 Signed-off-by: Danny Kopping <danny.kopping@grafana.com> * Add feature flag Signed-off-by: Danny Kopping <danny.kopping@grafana.com> * Refactor concurrency control Signed-off-by: Danny Kopping <danny.kopping@grafana.com> * Optimising dependencies/dependents funcs to not produce new slices each request Signed-off-by: Danny Kopping <danny.kopping@grafana.com> * Refactoring Signed-off-by: Danny Kopping <danny.kopping@grafana.com> * Rename flag Signed-off-by: Danny Kopping <danny.kopping@grafana.com> * Refactoring for performance, and to allow controller to be overridden Signed-off-by: Danny Kopping <danny.kopping@grafana.com> * Block until all rules, both sync & async, have completed evaluating Updated & added tests Review feedback nits Return empty map if not indeterminate Use highWatermark to track inflight requests counter Appease the linter Clarify feature flag Signed-off-by: Danny Kopping <danny.kopping@grafana.com> * Fix typo in CLI flag description Signed-off-by: Marco Pracucci <marco@pracucci.com> * Fixed auto-generated doc Signed-off-by: Marco Pracucci <marco@pracucci.com> * Improve doc Signed-off-by: Marco Pracucci <marco@pracucci.com> * Simplify the design to update concurrency controller once the rule evaluation has done Signed-off-by: Marco Pracucci <marco@pracucci.com> * Add more test cases to TestDependenciesEdgeCases Signed-off-by: Marco Pracucci <marco@pracucci.com> * Added more test cases to TestDependenciesEdgeCases Signed-off-by: Marco Pracucci <marco@pracucci.com> * Improved RuleConcurrencyController interface doc Signed-off-by: Marco Pracucci <marco@pracucci.com> * Introduced sequentialRuleEvalController Signed-off-by: Marco Pracucci <marco@pracucci.com> * Remove superfluous nil check in Group.metrics Signed-off-by: Marco Pracucci <marco@pracucci.com> * api: Serialize discovered and target labels into JSON directly (#13469) Converted maps into labels.Labels to avoid a lot of copying of data which leads to very high memory consumption while opening the /service-discovery endpoint in the Prometheus UI Signed-off-by: Leegin <114397475+Leegin-darknight@users.noreply.github.com> * api: Serialize discovered labels into JSON directly in dropped targets (#13484) Converted maps into labels.Labels to avoid a lot of copying of data which leads to very high memory consumption while opening the /service-discovery endpoint in the Prometheus UI Signed-off-by: Leegin <114397475+Leegin-darknight@users.noreply.github.com> * Add ShardedPostings() support to TSDB (#10421) This PR is a reference implementation of the proposal described in #10420. In addition to what described in #10420, in this PR I've introduced labels.StableHash(). The idea is to offer an hashing function which doesn't change over time, and that's used by query sharding in order to get a stable behaviour over time. The implementation of labels.StableHash() is the hashing function used by Prometheus before stringlabels, and what's used by Grafana Mimir for query sharding (because built before stringlabels was a thing). Follow up work As mentioned in #10420, if this PR is accepted I'm also open to upload another foundamental piece used by Grafana Mimir query sharding to accelerate the query execution: an optional, configurable and fast in-memory cache for the series hashes. Signed-off-by: Marco Pracucci <marco@pracucci.com> * storage/remote: document why two benchmarks are skipped One was silently doing nothing; one was doing something but the work didn't go up linearly with iteration count. Signed-off-by: Bryan Boreham <bjboreham@gmail.com> * Pod status changes not discovered by Kube Endpoints SD (#13337) * fix(discovery/kubernetes/endpoints): react to changes on Pods because some modifications can occur on them without triggering an update on the related Endpoints (The Pod phase changing from Pending to Running e.g.). --------- Signed-off-by: machine424 <ayoubmrini424@gmail.com> Co-authored-by: Guillermo Sanchez Gavier <gsanchez@newrelic.com> * Small improvements, add const, remove copypasta (#8106) Signed-off-by: Mikhail Fesenko <proggga@gmail.com> Signed-off-by: Jesus Vazquez <jesusvzpg@gmail.com> * Proposal to improve FPointSlice and HPointSlice allocation. (#13448) * Reusing points slice from previous series when the slice is under utilized * Adding comments on the bench test Signed-off-by: Alan Protasio <alanprot@gmail.com> * lint Signed-off-by: Nicolás Pazos <npazosmendez@gmail.com> * go mod tidy Signed-off-by: Nicolás Pazos <npazosmendez@gmail.com> --------- Signed-off-by: Julian Wiedmann <jwi@linux.ibm.com> Signed-off-by: Bryan Boreham <bjboreham@gmail.com> Signed-off-by: Erik Sommer <ersotech@posteo.de> Signed-off-by: Linas Medziunas <linas.medziunas@gmail.com> Signed-off-by: bwplotka <bwplotka@gmail.com> Signed-off-by: Arianna Vespri <arianna.vespri@yahoo.it> Signed-off-by: machine424 <ayoubmrini424@gmail.com> Signed-off-by: Daniel Nicholls <daniel.nicholls@resdiary.com> Signed-off-by: Daniel Kerbel <nmdanny@gmail.com> Signed-off-by: dependabot[bot] <support@github.com> Signed-off-by: Jan Fajerski <jfajersk@redhat.com> Signed-off-by: Kevin Mingtarja <kevin.mingtarja@gmail.com> Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> Signed-off-by: Marc Tuduri <marctc@protonmail.com> Signed-off-by: Paschalis Tsilias <tpaschalis@users.noreply.github.com> Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> Signed-off-by: Augustin Husson <augustin.husson@amadeus.com> Signed-off-by: Jeanette Tan <jeanette.tan@grafana.com> Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com> Signed-off-by: Kumar Kalpadiptya Roy <kalpadiptya.roy@outlook.com> Signed-off-by: Marco Pracucci <marco@pracucci.com> Signed-off-by: tyltr <tylitianrui@126.com> Signed-off-by: Ted Robertson 10043369+tredondo@users.noreply.github.com Signed-off-by: Ivan Babrou <github@ivan.computer> Signed-off-by: Arve Knudsen <arve.knudsen@gmail.com> Signed-off-by: Israel Blancas <iblancasa@gmail.com> Signed-off-by: Ziqi Zhao <zhaoziqi9146@gmail.com> Signed-off-by: Björn Rabenstein <github@rabenste.in> Signed-off-by: Goutham <gouthamve@gmail.com> Signed-off-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> Signed-off-by: Chris Marchbanks <csmarchbanks@gmail.com> Signed-off-by: Ben Ye <benye@amazon.com> Signed-off-by: Oleg Zaytsev <mail@olegzaytsev.com> Signed-off-by: SuperQ <superq@gmail.com> Signed-off-by: Ben Kochie <superq@gmail.com> Signed-off-by: Matthieu MOREL <matthieu.morel35@gmail.com> Signed-off-by: Paulin Todev <paulin.todev@gmail.com> Signed-off-by: Filip Petkovski <filip.petkovsky@gmail.com> Signed-off-by: beorn7 <beorn@grafana.com> Signed-off-by: Augustin Husson <husson.augustin@gmail.com> Signed-off-by: Yury Moladau <yurymolodov@gmail.com> Signed-off-by: Danny Kopping <danny.kopping@grafana.com> Signed-off-by: Leegin <114397475+Leegin-darknight@users.noreply.github.com> Signed-off-by: Mikhail Fesenko <proggga@gmail.com> Signed-off-by: Jesus Vazquez <jesusvzpg@gmail.com> Signed-off-by: Alan Protasio <alanprot@gmail.com> Signed-off-by: Nicolás Pazos <npazosmendez@gmail.com> Co-authored-by: Julian Wiedmann <jwi@linux.ibm.com> Co-authored-by: Bryan Boreham <bjboreham@gmail.com> Co-authored-by: Erik Sommer <ersotech@posteo.de> Co-authored-by: Linas Medziunas <linas.medziunas@gmail.com> Co-authored-by: Bartlomiej Plotka <bwplotka@gmail.com> Co-authored-by: Arianna Vespri <arianna.vespri@yahoo.it> Co-authored-by: machine424 <ayoubmrini424@gmail.com> Co-authored-by: daniel-resdiary <109083091+daniel-resdiary@users.noreply.github.com> Co-authored-by: Daniel Kerbel <nmdanny@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Jan Fajerski <jfajersk@redhat.com> Co-authored-by: Kevin Mingtarja <kevin.mingtarja@gmail.com> Co-authored-by: Paschalis Tsilias <tpaschalis@users.noreply.github.com> Co-authored-by: Marc Tudurí <marctc@protonmail.com> Co-authored-by: Paschalis Tsilias <paschalis.tsilias@grafana.com> Co-authored-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> Co-authored-by: Augustin Husson <husson.augustin@gmail.com> Co-authored-by: Björn Rabenstein <beorn@grafana.com> Co-authored-by: zenador <zenador@users.noreply.github.com> Co-authored-by: gotjosh <josue.abreu@gmail.com> Co-authored-by: Ben Kochie <superq@gmail.com> Co-authored-by: Kumar Kalpadiptya Roy <kalpadiptya.roy@outlook.com> Co-authored-by: Marco Pracucci <marco@pracucci.com> Co-authored-by: tyltr <tylitianrui@126.com> Co-authored-by: Ted Robertson <10043369+tredondo@users.noreply.github.com> Co-authored-by: Julien Pivotto <roidelapluie@o11y.eu> Co-authored-by: Matthias Loibl <mail@matthiasloibl.com> Co-authored-by: Ivan Babrou <github@ivan.computer> Co-authored-by: Arve Knudsen <arve.knudsen@gmail.com> Co-authored-by: Israel Blancas <iblancasa@gmail.com> Co-authored-by: Ziqi Zhao <zhaoziqi9146@gmail.com> Co-authored-by: George Krajcsovits <krajorama@users.noreply.github.com> Co-authored-by: Björn Rabenstein <github@rabenste.in> Co-authored-by: Goutham <gouthamve@gmail.com> Co-authored-by: Rewanth Tammana <22347290+rewanthtammana@users.noreply.github.com> Co-authored-by: Chris Marchbanks <csmarchbanks@gmail.com> Co-authored-by: Ben Ye <benye@amazon.com> Co-authored-by: Oleg Zaytsev <mail@olegzaytsev.com> Co-authored-by: Matthieu MOREL <matthieu.morel35@gmail.com> Co-authored-by: Paulin Todev <paulin.todev@gmail.com> Co-authored-by: Filip Petkovski <filip.petkovsky@gmail.com> Co-authored-by: Yury Molodov <yurymolodov@gmail.com> Co-authored-by: Danny Kopping <danny.kopping@grafana.com> Co-authored-by: Leegin <114397475+Leegin-darknight@users.noreply.github.com> Co-authored-by: Guillermo Sanchez Gavier <gsanchez@newrelic.com> Co-authored-by: Mikhail Fesenko <proggga@gmail.com> Co-authored-by: Alan Protasio <alanprot@gmail.com>
1534 lines
49 KiB
Go
1534 lines
49 KiB
Go
// Copyright 2021 The Prometheus Authors
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package tsdb
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
|
|
"github.com/go-kit/log/level"
|
|
|
|
"github.com/prometheus/prometheus/model/exemplar"
|
|
"github.com/prometheus/prometheus/model/histogram"
|
|
"github.com/prometheus/prometheus/model/labels"
|
|
"github.com/prometheus/prometheus/model/metadata"
|
|
"github.com/prometheus/prometheus/model/value"
|
|
"github.com/prometheus/prometheus/storage"
|
|
"github.com/prometheus/prometheus/tsdb/chunkenc"
|
|
"github.com/prometheus/prometheus/tsdb/chunks"
|
|
"github.com/prometheus/prometheus/tsdb/record"
|
|
)
|
|
|
|
// initAppender is a helper to initialize the time bounds of the head
|
|
// upon the first sample it receives.
|
|
type initAppender struct {
|
|
app storage.Appender
|
|
head *Head
|
|
}
|
|
|
|
var _ storage.GetRef = &initAppender{}
|
|
|
|
func (a *initAppender) Append(ref storage.SeriesRef, lset labels.Labels, t int64, v float64) (storage.SeriesRef, error) {
|
|
if a.app != nil {
|
|
return a.app.Append(ref, lset, t, v)
|
|
}
|
|
|
|
a.head.initTime(t)
|
|
a.app = a.head.appender()
|
|
return a.app.Append(ref, lset, t, v)
|
|
}
|
|
|
|
func (a *initAppender) AppendExemplar(ref storage.SeriesRef, l labels.Labels, e exemplar.Exemplar) (storage.SeriesRef, error) {
|
|
// Check if exemplar storage is enabled.
|
|
if !a.head.opts.EnableExemplarStorage || a.head.opts.MaxExemplars.Load() <= 0 {
|
|
return 0, nil
|
|
}
|
|
|
|
if a.app != nil {
|
|
return a.app.AppendExemplar(ref, l, e)
|
|
}
|
|
// We should never reach here given we would call Append before AppendExemplar
|
|
// and we probably want to always base head/WAL min time on sample times.
|
|
a.head.initTime(e.Ts)
|
|
a.app = a.head.appender()
|
|
|
|
return a.app.AppendExemplar(ref, l, e)
|
|
}
|
|
|
|
func (a *initAppender) AppendHistogram(ref storage.SeriesRef, l labels.Labels, t int64, h *histogram.Histogram, fh *histogram.FloatHistogram) (storage.SeriesRef, error) {
|
|
if a.app != nil {
|
|
return a.app.AppendHistogram(ref, l, t, h, fh)
|
|
}
|
|
a.head.initTime(t)
|
|
a.app = a.head.appender()
|
|
|
|
return a.app.AppendHistogram(ref, l, t, h, fh)
|
|
}
|
|
|
|
func (a *initAppender) UpdateMetadata(ref storage.SeriesRef, l labels.Labels, m metadata.Metadata) (storage.SeriesRef, error) {
|
|
if a.app != nil {
|
|
return a.app.UpdateMetadata(ref, l, m)
|
|
}
|
|
|
|
a.app = a.head.appender()
|
|
return a.app.UpdateMetadata(ref, l, m)
|
|
}
|
|
|
|
func (a *initAppender) AppendCTZeroSample(ref storage.SeriesRef, lset labels.Labels, t, ct int64) (storage.SeriesRef, error) {
|
|
if a.app != nil {
|
|
return a.app.AppendCTZeroSample(ref, lset, t, ct)
|
|
}
|
|
|
|
a.head.initTime(t)
|
|
a.app = a.head.appender()
|
|
|
|
return a.app.AppendCTZeroSample(ref, lset, t, ct)
|
|
}
|
|
|
|
// initTime initializes a head with the first timestamp. This only needs to be called
|
|
// for a completely fresh head with an empty WAL.
|
|
func (h *Head) initTime(t int64) {
|
|
if !h.minTime.CompareAndSwap(math.MaxInt64, t) {
|
|
return
|
|
}
|
|
// Ensure that max time is initialized to at least the min time we just set.
|
|
// Concurrent appenders may already have set it to a higher value.
|
|
h.maxTime.CompareAndSwap(math.MinInt64, t)
|
|
}
|
|
|
|
func (a *initAppender) GetRef(lset labels.Labels, hash uint64) (storage.SeriesRef, labels.Labels) {
|
|
if g, ok := a.app.(storage.GetRef); ok {
|
|
return g.GetRef(lset, hash)
|
|
}
|
|
return 0, labels.EmptyLabels()
|
|
}
|
|
|
|
func (a *initAppender) Commit() error {
|
|
if a.app == nil {
|
|
a.head.metrics.activeAppenders.Dec()
|
|
return nil
|
|
}
|
|
return a.app.Commit()
|
|
}
|
|
|
|
func (a *initAppender) Rollback() error {
|
|
if a.app == nil {
|
|
a.head.metrics.activeAppenders.Dec()
|
|
return nil
|
|
}
|
|
return a.app.Rollback()
|
|
}
|
|
|
|
// Appender returns a new Appender on the database.
|
|
func (h *Head) Appender(_ context.Context) storage.Appender {
|
|
h.metrics.activeAppenders.Inc()
|
|
|
|
// The head cache might not have a starting point yet. The init appender
|
|
// picks up the first appended timestamp as the base.
|
|
if h.MinTime() == math.MaxInt64 {
|
|
return &initAppender{
|
|
head: h,
|
|
}
|
|
}
|
|
return h.appender()
|
|
}
|
|
|
|
func (h *Head) appender() *headAppender {
|
|
minValidTime := h.appendableMinValidTime()
|
|
appendID, cleanupAppendIDsBelow := h.iso.newAppendID(minValidTime) // Every appender gets an ID that is cleared upon commit/rollback.
|
|
|
|
// Allocate the exemplars buffer only if exemplars are enabled.
|
|
var exemplarsBuf []exemplarWithSeriesRef
|
|
if h.opts.EnableExemplarStorage {
|
|
exemplarsBuf = h.getExemplarBuffer()
|
|
}
|
|
|
|
return &headAppender{
|
|
head: h,
|
|
minValidTime: minValidTime,
|
|
mint: math.MaxInt64,
|
|
maxt: math.MinInt64,
|
|
headMaxt: h.MaxTime(),
|
|
oooTimeWindow: h.opts.OutOfOrderTimeWindow.Load(),
|
|
samples: h.getAppendBuffer(),
|
|
sampleSeries: h.getSeriesBuffer(),
|
|
exemplars: exemplarsBuf,
|
|
histograms: h.getHistogramBuffer(),
|
|
floatHistograms: h.getFloatHistogramBuffer(),
|
|
metadata: h.getMetadataBuffer(),
|
|
appendID: appendID,
|
|
cleanupAppendIDsBelow: cleanupAppendIDsBelow,
|
|
}
|
|
}
|
|
|
|
// appendableMinValidTime returns the minimum valid timestamp for appends,
|
|
// such that samples stay ahead of prior blocks and the head compaction window.
|
|
func (h *Head) appendableMinValidTime() int64 {
|
|
// This boundary ensures that no samples will be added to the compaction window.
|
|
// This allows race-free, concurrent appending and compaction.
|
|
cwEnd := h.MaxTime() - h.chunkRange.Load()/2
|
|
|
|
// This boundary ensures that we avoid overlapping timeframes from one block to the next.
|
|
// While not necessary for correctness, it means we're not required to use vertical compaction.
|
|
minValid := h.minValidTime.Load()
|
|
|
|
return max(cwEnd, minValid)
|
|
}
|
|
|
|
// AppendableMinValidTime returns the minimum valid time for samples to be appended to the Head.
|
|
// Returns false if Head hasn't been initialized yet and the minimum time isn't known yet.
|
|
func (h *Head) AppendableMinValidTime() (int64, bool) {
|
|
if h.MinTime() == math.MaxInt64 {
|
|
return 0, false
|
|
}
|
|
|
|
return h.appendableMinValidTime(), true
|
|
}
|
|
|
|
func max(a, b int64) int64 {
|
|
if a > b {
|
|
return a
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (h *Head) getAppendBuffer() []record.RefSample {
|
|
b := h.appendPool.Get()
|
|
if b == nil {
|
|
return make([]record.RefSample, 0, 512)
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (h *Head) putAppendBuffer(b []record.RefSample) {
|
|
h.appendPool.Put(b[:0])
|
|
}
|
|
|
|
func (h *Head) getExemplarBuffer() []exemplarWithSeriesRef {
|
|
b := h.exemplarsPool.Get()
|
|
if b == nil {
|
|
return make([]exemplarWithSeriesRef, 0, 512)
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (h *Head) putExemplarBuffer(b []exemplarWithSeriesRef) {
|
|
if b == nil {
|
|
return
|
|
}
|
|
|
|
h.exemplarsPool.Put(b[:0])
|
|
}
|
|
|
|
func (h *Head) getHistogramBuffer() []record.RefHistogramSample {
|
|
b := h.histogramsPool.Get()
|
|
if b == nil {
|
|
return make([]record.RefHistogramSample, 0, 512)
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (h *Head) putHistogramBuffer(b []record.RefHistogramSample) {
|
|
h.histogramsPool.Put(b[:0])
|
|
}
|
|
|
|
func (h *Head) getFloatHistogramBuffer() []record.RefFloatHistogramSample {
|
|
b := h.floatHistogramsPool.Get()
|
|
if b == nil {
|
|
return make([]record.RefFloatHistogramSample, 0, 512)
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (h *Head) putFloatHistogramBuffer(b []record.RefFloatHistogramSample) {
|
|
h.floatHistogramsPool.Put(b[:0])
|
|
}
|
|
|
|
func (h *Head) getMetadataBuffer() []record.RefMetadata {
|
|
b := h.metadataPool.Get()
|
|
if b == nil {
|
|
return make([]record.RefMetadata, 0, 512)
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (h *Head) putMetadataBuffer(b []record.RefMetadata) {
|
|
h.metadataPool.Put(b[:0])
|
|
}
|
|
|
|
func (h *Head) getSeriesBuffer() []*memSeries {
|
|
b := h.seriesPool.Get()
|
|
if b == nil {
|
|
return make([]*memSeries, 0, 512)
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (h *Head) putSeriesBuffer(b []*memSeries) {
|
|
h.seriesPool.Put(b[:0])
|
|
}
|
|
|
|
func (h *Head) getBytesBuffer() []byte {
|
|
b := h.bytesPool.Get()
|
|
if b == nil {
|
|
return make([]byte, 0, 1024)
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (h *Head) putBytesBuffer(b []byte) {
|
|
h.bytesPool.Put(b[:0])
|
|
}
|
|
|
|
type exemplarWithSeriesRef struct {
|
|
ref storage.SeriesRef
|
|
exemplar exemplar.Exemplar
|
|
}
|
|
|
|
type headAppender struct {
|
|
head *Head
|
|
minValidTime int64 // No samples below this timestamp are allowed.
|
|
mint, maxt int64
|
|
headMaxt int64 // We track it here to not take the lock for every sample appended.
|
|
oooTimeWindow int64 // Use the same for the entire append, and don't load the atomic for each sample.
|
|
|
|
series []record.RefSeries // New series held by this appender.
|
|
samples []record.RefSample // New float samples held by this appender.
|
|
sampleSeries []*memSeries // Float series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once).
|
|
histograms []record.RefHistogramSample // New histogram samples held by this appender.
|
|
histogramSeries []*memSeries // HistogramSamples series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once).
|
|
floatHistograms []record.RefFloatHistogramSample // New float histogram samples held by this appender.
|
|
floatHistogramSeries []*memSeries // FloatHistogramSamples series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once).
|
|
metadata []record.RefMetadata // New metadata held by this appender.
|
|
metadataSeries []*memSeries // Series corresponding to the metadata held by this appender.
|
|
exemplars []exemplarWithSeriesRef // New exemplars held by this appender.
|
|
|
|
appendID, cleanupAppendIDsBelow uint64
|
|
closed bool
|
|
}
|
|
|
|
func (a *headAppender) Append(ref storage.SeriesRef, lset labels.Labels, t int64, v float64) (storage.SeriesRef, error) {
|
|
// For OOO inserts, this restriction is irrelevant and will be checked later once we confirm the sample is an in-order append.
|
|
// If OOO inserts are disabled, we may as well as check this as early as we can and avoid more work.
|
|
if a.oooTimeWindow == 0 && t < a.minValidTime {
|
|
a.head.metrics.outOfBoundSamples.WithLabelValues(sampleMetricTypeFloat).Inc()
|
|
return 0, storage.ErrOutOfBounds
|
|
}
|
|
|
|
s := a.head.series.getByID(chunks.HeadSeriesRef(ref))
|
|
if s == nil {
|
|
var err error
|
|
s, err = a.getOrCreate(lset)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
|
|
if value.IsStaleNaN(v) {
|
|
switch {
|
|
case s.lastHistogramValue != nil:
|
|
return a.AppendHistogram(ref, lset, t, &histogram.Histogram{Sum: v}, nil)
|
|
case s.lastFloatHistogramValue != nil:
|
|
return a.AppendHistogram(ref, lset, t, nil, &histogram.FloatHistogram{Sum: v})
|
|
}
|
|
}
|
|
|
|
s.Lock()
|
|
// TODO(codesome): If we definitely know at this point that the sample is ooo, then optimise
|
|
// to skip that sample from the WAL and write only in the WBL.
|
|
_, delta, err := s.appendable(t, v, a.headMaxt, a.minValidTime, a.oooTimeWindow)
|
|
if err == nil {
|
|
s.pendingCommit = true
|
|
}
|
|
s.Unlock()
|
|
if delta > 0 {
|
|
a.head.metrics.oooHistogram.Observe(float64(delta) / 1000)
|
|
}
|
|
if err != nil {
|
|
switch {
|
|
case errors.Is(err, storage.ErrOutOfOrderSample):
|
|
a.head.metrics.outOfOrderSamples.WithLabelValues(sampleMetricTypeFloat).Inc()
|
|
case errors.Is(err, storage.ErrTooOldSample):
|
|
a.head.metrics.tooOldSamples.WithLabelValues(sampleMetricTypeFloat).Inc()
|
|
}
|
|
return 0, err
|
|
}
|
|
|
|
if t < a.mint {
|
|
a.mint = t
|
|
}
|
|
if t > a.maxt {
|
|
a.maxt = t
|
|
}
|
|
|
|
a.samples = append(a.samples, record.RefSample{
|
|
Ref: s.ref,
|
|
T: t,
|
|
V: v,
|
|
})
|
|
a.sampleSeries = append(a.sampleSeries, s)
|
|
return storage.SeriesRef(s.ref), nil
|
|
}
|
|
|
|
// AppendCTZeroSample appends synthetic zero sample for ct timestamp. It returns
|
|
// error when sample can't be appended. See
|
|
// storage.CreatedTimestampAppender.AppendCTZeroSample for further documentation.
|
|
func (a *headAppender) AppendCTZeroSample(ref storage.SeriesRef, lset labels.Labels, t, ct int64) (storage.SeriesRef, error) {
|
|
if ct >= t {
|
|
return 0, fmt.Errorf("CT is newer or the same as sample's timestamp, ignoring")
|
|
}
|
|
|
|
s := a.head.series.getByID(chunks.HeadSeriesRef(ref))
|
|
if s == nil {
|
|
var err error
|
|
s, err = a.getOrCreate(lset)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
|
|
// Check if CT wouldn't be OOO vs samples we already might have for this series.
|
|
// NOTE(bwplotka): This will be often hit as it's expected for long living
|
|
// counters to share the same CT.
|
|
s.Lock()
|
|
isOOO, _, err := s.appendable(ct, 0, a.headMaxt, a.minValidTime, a.oooTimeWindow)
|
|
if err == nil {
|
|
s.pendingCommit = true
|
|
}
|
|
s.Unlock()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if isOOO {
|
|
return storage.SeriesRef(s.ref), storage.ErrOutOfOrderCT
|
|
}
|
|
|
|
if ct > a.maxt {
|
|
a.maxt = ct
|
|
}
|
|
a.samples = append(a.samples, record.RefSample{Ref: s.ref, T: ct, V: 0.0})
|
|
a.sampleSeries = append(a.sampleSeries, s)
|
|
return storage.SeriesRef(s.ref), nil
|
|
}
|
|
|
|
func (a *headAppender) getOrCreate(lset labels.Labels) (*memSeries, error) {
|
|
// Ensure no empty labels have gotten through.
|
|
lset = lset.WithoutEmpty()
|
|
if lset.IsEmpty() {
|
|
return nil, fmt.Errorf("empty labelset: %w", ErrInvalidSample)
|
|
}
|
|
if l, dup := lset.HasDuplicateLabelNames(); dup {
|
|
return nil, fmt.Errorf(`label name "%s" is not unique: %w`, l, ErrInvalidSample)
|
|
}
|
|
var created bool
|
|
var err error
|
|
s, created, err := a.head.getOrCreate(lset.Hash(), lset)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if created {
|
|
a.series = append(a.series, record.RefSeries{
|
|
Ref: s.ref,
|
|
Labels: lset,
|
|
})
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// appendable checks whether the given sample is valid for appending to the series. (if we return false and no error)
|
|
// The sample belongs to the out of order chunk if we return true and no error.
|
|
// An error signifies the sample cannot be handled.
|
|
func (s *memSeries) appendable(t int64, v float64, headMaxt, minValidTime, oooTimeWindow int64) (isOOO bool, oooDelta int64, err error) {
|
|
// Check if we can append in the in-order chunk.
|
|
if t >= minValidTime {
|
|
if s.headChunks == nil {
|
|
// The series has no sample and was freshly created.
|
|
return false, 0, nil
|
|
}
|
|
msMaxt := s.maxTime()
|
|
if t > msMaxt {
|
|
return false, 0, nil
|
|
}
|
|
if t == msMaxt {
|
|
// We are allowing exact duplicates as we can encounter them in valid cases
|
|
// like federation and erroring out at that time would be extremely noisy.
|
|
// This only checks against the latest in-order sample.
|
|
// The OOO headchunk has its own method to detect these duplicates.
|
|
if math.Float64bits(s.lastValue) != math.Float64bits(v) {
|
|
return false, 0, storage.ErrDuplicateSampleForTimestamp
|
|
}
|
|
// Sample is identical (ts + value) with most current (highest ts) sample in sampleBuf.
|
|
return false, 0, nil
|
|
}
|
|
}
|
|
|
|
// The sample cannot go in the in-order chunk. Check if it can go in the out-of-order chunk.
|
|
if oooTimeWindow > 0 && t >= headMaxt-oooTimeWindow {
|
|
return true, headMaxt - t, nil
|
|
}
|
|
|
|
// The sample cannot go in both in-order and out-of-order chunk.
|
|
if oooTimeWindow > 0 {
|
|
return true, headMaxt - t, storage.ErrTooOldSample
|
|
}
|
|
if t < minValidTime {
|
|
return false, headMaxt - t, storage.ErrOutOfBounds
|
|
}
|
|
return false, headMaxt - t, storage.ErrOutOfOrderSample
|
|
}
|
|
|
|
// appendableHistogram checks whether the given histogram is valid for appending to the series.
|
|
func (s *memSeries) appendableHistogram(t int64, h *histogram.Histogram) error {
|
|
if s.headChunks == nil {
|
|
return nil
|
|
}
|
|
|
|
if t > s.headChunks.maxTime {
|
|
return nil
|
|
}
|
|
if t < s.headChunks.maxTime {
|
|
return storage.ErrOutOfOrderSample
|
|
}
|
|
|
|
// We are allowing exact duplicates as we can encounter them in valid cases
|
|
// like federation and erroring out at that time would be extremely noisy.
|
|
if !h.Equals(s.lastHistogramValue) {
|
|
return storage.ErrDuplicateSampleForTimestamp
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// appendableFloatHistogram checks whether the given float histogram is valid for appending to the series.
|
|
func (s *memSeries) appendableFloatHistogram(t int64, fh *histogram.FloatHistogram) error {
|
|
if s.headChunks == nil {
|
|
return nil
|
|
}
|
|
|
|
if t > s.headChunks.maxTime {
|
|
return nil
|
|
}
|
|
if t < s.headChunks.maxTime {
|
|
return storage.ErrOutOfOrderSample
|
|
}
|
|
|
|
// We are allowing exact duplicates as we can encounter them in valid cases
|
|
// like federation and erroring out at that time would be extremely noisy.
|
|
if !fh.Equals(s.lastFloatHistogramValue) {
|
|
return storage.ErrDuplicateSampleForTimestamp
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// AppendExemplar for headAppender assumes the series ref already exists, and so it doesn't
|
|
// use getOrCreate or make any of the lset validity checks that Append does.
|
|
func (a *headAppender) AppendExemplar(ref storage.SeriesRef, lset labels.Labels, e exemplar.Exemplar) (storage.SeriesRef, error) {
|
|
// Check if exemplar storage is enabled.
|
|
if !a.head.opts.EnableExemplarStorage || a.head.opts.MaxExemplars.Load() <= 0 {
|
|
return 0, nil
|
|
}
|
|
|
|
// Get Series
|
|
s := a.head.series.getByID(chunks.HeadSeriesRef(ref))
|
|
if s == nil {
|
|
s = a.head.series.getByHash(lset.Hash(), lset)
|
|
if s != nil {
|
|
ref = storage.SeriesRef(s.ref)
|
|
}
|
|
}
|
|
if s == nil {
|
|
return 0, fmt.Errorf("unknown HeadSeriesRef when trying to add exemplar: %d", ref)
|
|
}
|
|
|
|
// Ensure no empty labels have gotten through.
|
|
e.Labels = e.Labels.WithoutEmpty()
|
|
|
|
err := a.head.exemplars.ValidateExemplar(s.lset, e)
|
|
if err != nil {
|
|
if errors.Is(err, storage.ErrDuplicateExemplar) || errors.Is(err, storage.ErrExemplarsDisabled) {
|
|
// Duplicate, don't return an error but don't accept the exemplar.
|
|
return 0, nil
|
|
}
|
|
return 0, err
|
|
}
|
|
|
|
a.exemplars = append(a.exemplars, exemplarWithSeriesRef{ref, e})
|
|
|
|
return storage.SeriesRef(s.ref), nil
|
|
}
|
|
|
|
func (a *headAppender) AppendHistogram(ref storage.SeriesRef, lset labels.Labels, t int64, h *histogram.Histogram, fh *histogram.FloatHistogram) (storage.SeriesRef, error) {
|
|
if !a.head.opts.EnableNativeHistograms.Load() {
|
|
return 0, storage.ErrNativeHistogramsDisabled
|
|
}
|
|
|
|
if t < a.minValidTime {
|
|
a.head.metrics.outOfBoundSamples.WithLabelValues(sampleMetricTypeHistogram).Inc()
|
|
return 0, storage.ErrOutOfBounds
|
|
}
|
|
|
|
if h != nil {
|
|
if err := h.Validate(); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
|
|
if fh != nil {
|
|
if err := fh.Validate(); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
|
|
s := a.head.series.getByID(chunks.HeadSeriesRef(ref))
|
|
if s == nil {
|
|
// Ensure no empty labels have gotten through.
|
|
lset = lset.WithoutEmpty()
|
|
if lset.IsEmpty() {
|
|
return 0, fmt.Errorf("empty labelset: %w", ErrInvalidSample)
|
|
}
|
|
|
|
if l, dup := lset.HasDuplicateLabelNames(); dup {
|
|
return 0, fmt.Errorf(`label name "%s" is not unique: %w`, l, ErrInvalidSample)
|
|
}
|
|
|
|
var created bool
|
|
var err error
|
|
s, created, err = a.head.getOrCreate(lset.Hash(), lset)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if created {
|
|
switch {
|
|
case h != nil:
|
|
s.lastHistogramValue = &histogram.Histogram{}
|
|
case fh != nil:
|
|
s.lastFloatHistogramValue = &histogram.FloatHistogram{}
|
|
}
|
|
a.series = append(a.series, record.RefSeries{
|
|
Ref: s.ref,
|
|
Labels: lset,
|
|
})
|
|
}
|
|
}
|
|
|
|
switch {
|
|
case h != nil:
|
|
s.Lock()
|
|
if err := s.appendableHistogram(t, h); err != nil {
|
|
s.Unlock()
|
|
if errors.Is(err, storage.ErrOutOfOrderSample) {
|
|
a.head.metrics.outOfOrderSamples.WithLabelValues(sampleMetricTypeHistogram).Inc()
|
|
}
|
|
return 0, err
|
|
}
|
|
s.pendingCommit = true
|
|
s.Unlock()
|
|
a.histograms = append(a.histograms, record.RefHistogramSample{
|
|
Ref: s.ref,
|
|
T: t,
|
|
H: h,
|
|
})
|
|
a.histogramSeries = append(a.histogramSeries, s)
|
|
case fh != nil:
|
|
s.Lock()
|
|
if err := s.appendableFloatHistogram(t, fh); err != nil {
|
|
s.Unlock()
|
|
if errors.Is(err, storage.ErrOutOfOrderSample) {
|
|
a.head.metrics.outOfOrderSamples.WithLabelValues(sampleMetricTypeHistogram).Inc()
|
|
}
|
|
return 0, err
|
|
}
|
|
s.pendingCommit = true
|
|
s.Unlock()
|
|
a.floatHistograms = append(a.floatHistograms, record.RefFloatHistogramSample{
|
|
Ref: s.ref,
|
|
T: t,
|
|
FH: fh,
|
|
})
|
|
a.floatHistogramSeries = append(a.floatHistogramSeries, s)
|
|
}
|
|
|
|
if t < a.mint {
|
|
a.mint = t
|
|
}
|
|
if t > a.maxt {
|
|
a.maxt = t
|
|
}
|
|
|
|
return storage.SeriesRef(s.ref), nil
|
|
}
|
|
|
|
// UpdateMetadata for headAppender assumes the series ref already exists, and so it doesn't
|
|
// use getOrCreate or make any of the lset sanity checks that Append does.
|
|
func (a *headAppender) UpdateMetadata(ref storage.SeriesRef, lset labels.Labels, meta metadata.Metadata) (storage.SeriesRef, error) {
|
|
s := a.head.series.getByID(chunks.HeadSeriesRef(ref))
|
|
if s == nil {
|
|
s = a.head.series.getByHash(lset.Hash(), lset)
|
|
if s != nil {
|
|
ref = storage.SeriesRef(s.ref)
|
|
}
|
|
}
|
|
if s == nil {
|
|
return 0, fmt.Errorf("unknown series when trying to add metadata with HeadSeriesRef: %d and labels: %s", ref, lset)
|
|
}
|
|
|
|
s.Lock()
|
|
hasNewMetadata := s.meta == nil || *s.meta != meta
|
|
s.Unlock()
|
|
|
|
if hasNewMetadata {
|
|
a.metadata = append(a.metadata, record.RefMetadata{
|
|
Ref: s.ref,
|
|
Type: record.GetMetricType(meta.Type),
|
|
Unit: meta.Unit,
|
|
Help: meta.Help,
|
|
})
|
|
a.metadataSeries = append(a.metadataSeries, s)
|
|
}
|
|
|
|
return ref, nil
|
|
}
|
|
|
|
var _ storage.GetRef = &headAppender{}
|
|
|
|
func (a *headAppender) GetRef(lset labels.Labels, hash uint64) (storage.SeriesRef, labels.Labels) {
|
|
s := a.head.series.getByHash(hash, lset)
|
|
if s == nil {
|
|
return 0, labels.EmptyLabels()
|
|
}
|
|
// returned labels must be suitable to pass to Append()
|
|
return storage.SeriesRef(s.ref), s.lset
|
|
}
|
|
|
|
// log writes all headAppender's data to the WAL.
|
|
func (a *headAppender) log() error {
|
|
if a.head.wal == nil {
|
|
return nil
|
|
}
|
|
|
|
buf := a.head.getBytesBuffer()
|
|
defer func() { a.head.putBytesBuffer(buf) }()
|
|
|
|
var rec []byte
|
|
var enc record.Encoder
|
|
|
|
if len(a.series) > 0 {
|
|
rec = enc.Series(a.series, buf)
|
|
buf = rec[:0]
|
|
|
|
if err := a.head.wal.Log(rec); err != nil {
|
|
return fmt.Errorf("log series: %w", err)
|
|
}
|
|
}
|
|
if len(a.metadata) > 0 {
|
|
rec = enc.Metadata(a.metadata, buf)
|
|
buf = rec[:0]
|
|
|
|
if err := a.head.wal.Log(rec); err != nil {
|
|
return fmt.Errorf("log metadata: %w", err)
|
|
}
|
|
}
|
|
if len(a.samples) > 0 {
|
|
rec = enc.Samples(a.samples, buf)
|
|
buf = rec[:0]
|
|
|
|
if err := a.head.wal.Log(rec); err != nil {
|
|
return fmt.Errorf("log samples: %w", err)
|
|
}
|
|
}
|
|
if len(a.histograms) > 0 {
|
|
rec = enc.HistogramSamples(a.histograms, buf)
|
|
buf = rec[:0]
|
|
if err := a.head.wal.Log(rec); err != nil {
|
|
return fmt.Errorf("log histograms: %w", err)
|
|
}
|
|
}
|
|
if len(a.floatHistograms) > 0 {
|
|
rec = enc.FloatHistogramSamples(a.floatHistograms, buf)
|
|
buf = rec[:0]
|
|
if err := a.head.wal.Log(rec); err != nil {
|
|
return fmt.Errorf("log float histograms: %w", err)
|
|
}
|
|
}
|
|
// Exemplars should be logged after samples (float/native histogram/etc),
|
|
// otherwise it might happen that we send the exemplars in a remote write
|
|
// batch before the samples, which in turn means the exemplar is rejected
|
|
// for missing series, since series are created due to samples.
|
|
if len(a.exemplars) > 0 {
|
|
rec = enc.Exemplars(exemplarsForEncoding(a.exemplars), buf)
|
|
buf = rec[:0]
|
|
|
|
if err := a.head.wal.Log(rec); err != nil {
|
|
return fmt.Errorf("log exemplars: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func exemplarsForEncoding(es []exemplarWithSeriesRef) []record.RefExemplar {
|
|
ret := make([]record.RefExemplar, 0, len(es))
|
|
for _, e := range es {
|
|
ret = append(ret, record.RefExemplar{
|
|
Ref: chunks.HeadSeriesRef(e.ref),
|
|
T: e.exemplar.Ts,
|
|
V: e.exemplar.Value,
|
|
Labels: e.exemplar.Labels,
|
|
})
|
|
}
|
|
return ret
|
|
}
|
|
|
|
// Commit writes to the WAL and adds the data to the Head.
|
|
// TODO(codesome): Refactor this method to reduce indentation and make it more readable.
|
|
func (a *headAppender) Commit() (err error) {
|
|
if a.closed {
|
|
return ErrAppenderClosed
|
|
}
|
|
defer func() { a.closed = true }()
|
|
|
|
if err := a.log(); err != nil {
|
|
_ = a.Rollback() // Most likely the same error will happen again.
|
|
return fmt.Errorf("write to WAL: %w", err)
|
|
}
|
|
|
|
if a.head.writeNotified != nil {
|
|
a.head.writeNotified.Notify()
|
|
}
|
|
|
|
// No errors logging to WAL, so pass the exemplars along to the in memory storage.
|
|
for _, e := range a.exemplars {
|
|
s := a.head.series.getByID(chunks.HeadSeriesRef(e.ref))
|
|
if s == nil {
|
|
// This is very unlikely to happen, but we have seen it in the wild.
|
|
// It means that the series was truncated between AppendExemplar and Commit.
|
|
// See TestHeadCompactionWhileAppendAndCommitExemplar.
|
|
continue
|
|
}
|
|
// We don't instrument exemplar appends here, all is instrumented by storage.
|
|
if err := a.head.exemplars.AddExemplar(s.lset, e.exemplar); err != nil {
|
|
if errors.Is(err, storage.ErrOutOfOrderExemplar) {
|
|
continue
|
|
}
|
|
level.Debug(a.head.logger).Log("msg", "Unknown error while adding exemplar", "err", err)
|
|
}
|
|
}
|
|
|
|
defer a.head.metrics.activeAppenders.Dec()
|
|
defer a.head.putAppendBuffer(a.samples)
|
|
defer a.head.putSeriesBuffer(a.sampleSeries)
|
|
defer a.head.putExemplarBuffer(a.exemplars)
|
|
defer a.head.putHistogramBuffer(a.histograms)
|
|
defer a.head.putFloatHistogramBuffer(a.floatHistograms)
|
|
defer a.head.putMetadataBuffer(a.metadata)
|
|
defer a.head.iso.closeAppend(a.appendID)
|
|
|
|
var (
|
|
samplesAppended = len(a.samples)
|
|
oooAccepted int // number of samples out of order but accepted: with ooo enabled and within time window
|
|
oooRejected int // number of samples rejected due to: out of order but OOO support disabled.
|
|
tooOldRejected int // number of samples rejected due to: that are out of order but too old (OOO support enabled, but outside time window)
|
|
oobRejected int // number of samples rejected due to: out of bounds: with t < minValidTime (OOO support disabled)
|
|
inOrderMint int64 = math.MaxInt64
|
|
inOrderMaxt int64 = math.MinInt64
|
|
ooomint int64 = math.MaxInt64
|
|
ooomaxt int64 = math.MinInt64
|
|
wblSamples []record.RefSample
|
|
oooMmapMarkers map[chunks.HeadSeriesRef]chunks.ChunkDiskMapperRef
|
|
oooRecords [][]byte
|
|
oooCapMax = a.head.opts.OutOfOrderCapMax.Load()
|
|
series *memSeries
|
|
appendChunkOpts = chunkOpts{
|
|
chunkDiskMapper: a.head.chunkDiskMapper,
|
|
chunkRange: a.head.chunkRange.Load(),
|
|
samplesPerChunk: a.head.opts.SamplesPerChunk,
|
|
}
|
|
enc record.Encoder
|
|
)
|
|
defer func() {
|
|
for i := range oooRecords {
|
|
a.head.putBytesBuffer(oooRecords[i][:0])
|
|
}
|
|
}()
|
|
collectOOORecords := func() {
|
|
if a.head.wbl == nil {
|
|
// WBL is not enabled. So no need to collect.
|
|
wblSamples = nil
|
|
oooMmapMarkers = nil
|
|
return
|
|
}
|
|
// The m-map happens before adding a new sample. So we collect
|
|
// the m-map markers first, and then samples.
|
|
// WBL Graphically:
|
|
// WBL Before this Commit(): [old samples before this commit for chunk 1]
|
|
// WBL After this Commit(): [old samples before this commit for chunk 1][new samples in this commit for chunk 1]mmapmarker1[samples for chunk 2]mmapmarker2[samples for chunk 3]
|
|
if oooMmapMarkers != nil {
|
|
markers := make([]record.RefMmapMarker, 0, len(oooMmapMarkers))
|
|
for ref, mmapRef := range oooMmapMarkers {
|
|
markers = append(markers, record.RefMmapMarker{
|
|
Ref: ref,
|
|
MmapRef: mmapRef,
|
|
})
|
|
}
|
|
r := enc.MmapMarkers(markers, a.head.getBytesBuffer())
|
|
oooRecords = append(oooRecords, r)
|
|
}
|
|
|
|
if len(wblSamples) > 0 {
|
|
r := enc.Samples(wblSamples, a.head.getBytesBuffer())
|
|
oooRecords = append(oooRecords, r)
|
|
}
|
|
|
|
wblSamples = nil
|
|
oooMmapMarkers = nil
|
|
}
|
|
for i, s := range a.samples {
|
|
series = a.sampleSeries[i]
|
|
series.Lock()
|
|
|
|
oooSample, _, err := series.appendable(s.T, s.V, a.headMaxt, a.minValidTime, a.oooTimeWindow)
|
|
switch {
|
|
case err == nil:
|
|
// Do nothing.
|
|
case errors.Is(err, storage.ErrOutOfOrderSample):
|
|
samplesAppended--
|
|
oooRejected++
|
|
case errors.Is(err, storage.ErrOutOfBounds):
|
|
samplesAppended--
|
|
oobRejected++
|
|
case errors.Is(err, storage.ErrTooOldSample):
|
|
samplesAppended--
|
|
tooOldRejected++
|
|
default:
|
|
samplesAppended--
|
|
}
|
|
|
|
var ok, chunkCreated bool
|
|
|
|
switch {
|
|
case err != nil:
|
|
// Do nothing here.
|
|
case oooSample:
|
|
// Sample is OOO and OOO handling is enabled
|
|
// and the delta is within the OOO tolerance.
|
|
var mmapRef chunks.ChunkDiskMapperRef
|
|
ok, chunkCreated, mmapRef = series.insert(s.T, s.V, a.head.chunkDiskMapper, oooCapMax)
|
|
if chunkCreated {
|
|
r, ok := oooMmapMarkers[series.ref]
|
|
if !ok || r != 0 {
|
|
// !ok means there are no markers collected for these samples yet. So we first flush the samples
|
|
// before setting this m-map marker.
|
|
|
|
// r != 0 means we have already m-mapped a chunk for this series in the same Commit().
|
|
// Hence, before we m-map again, we should add the samples and m-map markers
|
|
// seen till now to the WBL records.
|
|
collectOOORecords()
|
|
}
|
|
|
|
if oooMmapMarkers == nil {
|
|
oooMmapMarkers = make(map[chunks.HeadSeriesRef]chunks.ChunkDiskMapperRef)
|
|
}
|
|
oooMmapMarkers[series.ref] = mmapRef
|
|
}
|
|
if ok {
|
|
wblSamples = append(wblSamples, s)
|
|
if s.T < ooomint {
|
|
ooomint = s.T
|
|
}
|
|
if s.T > ooomaxt {
|
|
ooomaxt = s.T
|
|
}
|
|
oooAccepted++
|
|
} else {
|
|
// Sample is an exact duplicate of the last sample.
|
|
// NOTE: We can only detect updates if they clash with a sample in the OOOHeadChunk,
|
|
// not with samples in already flushed OOO chunks.
|
|
// TODO(codesome): Add error reporting? It depends on addressing https://github.com/prometheus/prometheus/discussions/10305.
|
|
samplesAppended--
|
|
}
|
|
default:
|
|
ok, chunkCreated = series.append(s.T, s.V, a.appendID, appendChunkOpts)
|
|
if ok {
|
|
if s.T < inOrderMint {
|
|
inOrderMint = s.T
|
|
}
|
|
if s.T > inOrderMaxt {
|
|
inOrderMaxt = s.T
|
|
}
|
|
} else {
|
|
// The sample is an exact duplicate, and should be silently dropped.
|
|
samplesAppended--
|
|
}
|
|
}
|
|
|
|
if chunkCreated {
|
|
a.head.metrics.chunks.Inc()
|
|
a.head.metrics.chunksCreated.Inc()
|
|
}
|
|
|
|
series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow)
|
|
series.pendingCommit = false
|
|
series.Unlock()
|
|
}
|
|
|
|
histogramsTotal := len(a.histograms)
|
|
histoOOORejected := 0
|
|
for i, s := range a.histograms {
|
|
series = a.histogramSeries[i]
|
|
series.Lock()
|
|
ok, chunkCreated := series.appendHistogram(s.T, s.H, a.appendID, appendChunkOpts)
|
|
series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow)
|
|
series.pendingCommit = false
|
|
series.Unlock()
|
|
|
|
if ok {
|
|
if s.T < inOrderMint {
|
|
inOrderMint = s.T
|
|
}
|
|
if s.T > inOrderMaxt {
|
|
inOrderMaxt = s.T
|
|
}
|
|
} else {
|
|
histogramsTotal--
|
|
histoOOORejected++
|
|
}
|
|
if chunkCreated {
|
|
a.head.metrics.chunks.Inc()
|
|
a.head.metrics.chunksCreated.Inc()
|
|
}
|
|
}
|
|
|
|
histogramsTotal += len(a.floatHistograms)
|
|
for i, s := range a.floatHistograms {
|
|
series = a.floatHistogramSeries[i]
|
|
series.Lock()
|
|
ok, chunkCreated := series.appendFloatHistogram(s.T, s.FH, a.appendID, appendChunkOpts)
|
|
series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow)
|
|
series.pendingCommit = false
|
|
series.Unlock()
|
|
|
|
if ok {
|
|
if s.T < inOrderMint {
|
|
inOrderMint = s.T
|
|
}
|
|
if s.T > inOrderMaxt {
|
|
inOrderMaxt = s.T
|
|
}
|
|
} else {
|
|
histogramsTotal--
|
|
histoOOORejected++
|
|
}
|
|
if chunkCreated {
|
|
a.head.metrics.chunks.Inc()
|
|
a.head.metrics.chunksCreated.Inc()
|
|
}
|
|
}
|
|
|
|
for i, m := range a.metadata {
|
|
series = a.metadataSeries[i]
|
|
series.Lock()
|
|
series.meta = &metadata.Metadata{Type: record.ToMetricType(m.Type), Unit: m.Unit, Help: m.Help}
|
|
series.Unlock()
|
|
}
|
|
|
|
a.head.metrics.outOfOrderSamples.WithLabelValues(sampleMetricTypeFloat).Add(float64(oooRejected))
|
|
a.head.metrics.outOfOrderSamples.WithLabelValues(sampleMetricTypeHistogram).Add(float64(histoOOORejected))
|
|
a.head.metrics.outOfBoundSamples.WithLabelValues(sampleMetricTypeFloat).Add(float64(oobRejected))
|
|
a.head.metrics.tooOldSamples.WithLabelValues(sampleMetricTypeFloat).Add(float64(tooOldRejected))
|
|
a.head.metrics.samplesAppended.WithLabelValues(sampleMetricTypeFloat).Add(float64(samplesAppended))
|
|
a.head.metrics.samplesAppended.WithLabelValues(sampleMetricTypeHistogram).Add(float64(histogramsTotal))
|
|
a.head.metrics.outOfOrderSamplesAppended.Add(float64(oooAccepted))
|
|
a.head.updateMinMaxTime(inOrderMint, inOrderMaxt)
|
|
a.head.updateMinOOOMaxOOOTime(ooomint, ooomaxt)
|
|
|
|
collectOOORecords()
|
|
if a.head.wbl != nil {
|
|
if err := a.head.wbl.Log(oooRecords...); err != nil {
|
|
// TODO(codesome): Currently WBL logging of ooo samples is best effort here since we cannot try logging
|
|
// until we have found what samples become OOO. We can try having a metric for this failure.
|
|
// Returning the error here is not correct because we have already put the samples into the memory,
|
|
// hence the append/insert was a success.
|
|
level.Error(a.head.logger).Log("msg", "Failed to log out of order samples into the WAL", "err", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// insert is like append, except it inserts. Used for OOO samples.
|
|
func (s *memSeries) insert(t int64, v float64, chunkDiskMapper *chunks.ChunkDiskMapper, oooCapMax int64) (inserted, chunkCreated bool, mmapRef chunks.ChunkDiskMapperRef) {
|
|
if s.ooo == nil {
|
|
s.ooo = &memSeriesOOOFields{}
|
|
}
|
|
c := s.ooo.oooHeadChunk
|
|
if c == nil || c.chunk.NumSamples() == int(oooCapMax) {
|
|
// Note: If no new samples come in then we rely on compaction to clean up stale in-memory OOO chunks.
|
|
c, mmapRef = s.cutNewOOOHeadChunk(t, chunkDiskMapper)
|
|
chunkCreated = true
|
|
}
|
|
|
|
ok := c.chunk.Insert(t, v)
|
|
if ok {
|
|
if chunkCreated || t < c.minTime {
|
|
c.minTime = t
|
|
}
|
|
if chunkCreated || t > c.maxTime {
|
|
c.maxTime = t
|
|
}
|
|
}
|
|
return ok, chunkCreated, mmapRef
|
|
}
|
|
|
|
// chunkOpts are chunk-level options that are passed when appending to a memSeries.
|
|
type chunkOpts struct {
|
|
chunkDiskMapper *chunks.ChunkDiskMapper
|
|
chunkRange int64
|
|
samplesPerChunk int
|
|
}
|
|
|
|
// append adds the sample (t, v) to the series. The caller also has to provide
|
|
// the appendID for isolation. (The appendID can be zero, which results in no
|
|
// isolation for this append.)
|
|
// It is unsafe to call this concurrently with s.iterator(...) without holding the series lock.
|
|
func (s *memSeries) append(t int64, v float64, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
|
|
c, sampleInOrder, chunkCreated := s.appendPreprocessor(t, chunkenc.EncXOR, o)
|
|
if !sampleInOrder {
|
|
return sampleInOrder, chunkCreated
|
|
}
|
|
s.app.Append(t, v)
|
|
|
|
c.maxTime = t
|
|
|
|
s.lastValue = v
|
|
s.lastHistogramValue = nil
|
|
s.lastFloatHistogramValue = nil
|
|
|
|
if appendID > 0 {
|
|
s.txs.add(appendID)
|
|
}
|
|
|
|
return true, chunkCreated
|
|
}
|
|
|
|
// appendHistogram adds the histogram.
|
|
// It is unsafe to call this concurrently with s.iterator(...) without holding the series lock.
|
|
// In case of recoding the existing chunk, a new chunk is allocated and the old chunk is dropped.
|
|
// To keep the meaning of prometheus_tsdb_head_chunks and prometheus_tsdb_head_chunks_created_total
|
|
// consistent, we return chunkCreated=false in this case.
|
|
func (s *memSeries) appendHistogram(t int64, h *histogram.Histogram, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
|
|
// Head controls the execution of recoding, so that we own the proper
|
|
// chunk reference afterwards and mmap used up chunks.
|
|
|
|
// Ignoring ok is ok, since we don't want to compare to the wrong previous appender anyway.
|
|
prevApp, _ := s.app.(*chunkenc.HistogramAppender)
|
|
|
|
c, sampleInOrder, chunkCreated := s.histogramsAppendPreprocessor(t, chunkenc.EncHistogram, o)
|
|
if !sampleInOrder {
|
|
return sampleInOrder, chunkCreated
|
|
}
|
|
|
|
var (
|
|
newChunk chunkenc.Chunk
|
|
recoded bool
|
|
)
|
|
|
|
if !chunkCreated {
|
|
// Ignore the previous appender if we continue the current chunk.
|
|
prevApp = nil
|
|
}
|
|
|
|
newChunk, recoded, s.app, _ = s.app.AppendHistogram(prevApp, t, h, false) // false=request a new chunk if needed
|
|
|
|
s.lastHistogramValue = h
|
|
s.lastFloatHistogramValue = nil
|
|
|
|
if appendID > 0 {
|
|
s.txs.add(appendID)
|
|
}
|
|
|
|
if newChunk == nil { // Sample was appended to existing chunk or is the first sample in a new chunk.
|
|
c.maxTime = t
|
|
return true, chunkCreated
|
|
}
|
|
|
|
if recoded { // The appender needed to recode the chunk.
|
|
c.maxTime = t
|
|
c.chunk = newChunk
|
|
return true, false
|
|
}
|
|
|
|
s.headChunks = &memChunk{
|
|
chunk: newChunk,
|
|
minTime: t,
|
|
maxTime: t,
|
|
prev: s.headChunks,
|
|
}
|
|
s.nextAt = rangeForTimestamp(t, o.chunkRange)
|
|
return true, true
|
|
}
|
|
|
|
// appendFloatHistogram adds the float histogram.
|
|
// It is unsafe to call this concurrently with s.iterator(...) without holding the series lock.
|
|
// In case of recoding the existing chunk, a new chunk is allocated and the old chunk is dropped.
|
|
// To keep the meaning of prometheus_tsdb_head_chunks and prometheus_tsdb_head_chunks_created_total
|
|
// consistent, we return chunkCreated=false in this case.
|
|
func (s *memSeries) appendFloatHistogram(t int64, fh *histogram.FloatHistogram, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
|
|
// Head controls the execution of recoding, so that we own the proper
|
|
// chunk reference afterwards and mmap used up chunks.
|
|
|
|
// Ignoring ok is ok, since we don't want to compare to the wrong previous appender anyway.
|
|
prevApp, _ := s.app.(*chunkenc.FloatHistogramAppender)
|
|
|
|
c, sampleInOrder, chunkCreated := s.histogramsAppendPreprocessor(t, chunkenc.EncFloatHistogram, o)
|
|
if !sampleInOrder {
|
|
return sampleInOrder, chunkCreated
|
|
}
|
|
|
|
var (
|
|
newChunk chunkenc.Chunk
|
|
recoded bool
|
|
)
|
|
|
|
if !chunkCreated {
|
|
// Ignore the previous appender if we continue the current chunk.
|
|
prevApp = nil
|
|
}
|
|
|
|
newChunk, recoded, s.app, _ = s.app.AppendFloatHistogram(prevApp, t, fh, false) // False means request a new chunk if needed.
|
|
|
|
s.lastHistogramValue = nil
|
|
s.lastFloatHistogramValue = fh
|
|
|
|
if appendID > 0 {
|
|
s.txs.add(appendID)
|
|
}
|
|
|
|
if newChunk == nil { // Sample was appended to existing chunk or is the first sample in a new chunk.
|
|
c.maxTime = t
|
|
return true, chunkCreated
|
|
}
|
|
|
|
if recoded { // The appender needed to recode the chunk.
|
|
c.maxTime = t
|
|
c.chunk = newChunk
|
|
return true, false
|
|
}
|
|
|
|
s.headChunks = &memChunk{
|
|
chunk: newChunk,
|
|
minTime: t,
|
|
maxTime: t,
|
|
prev: s.headChunks,
|
|
}
|
|
s.nextAt = rangeForTimestamp(t, o.chunkRange)
|
|
return true, true
|
|
}
|
|
|
|
// appendPreprocessor takes care of cutting new XOR chunks and m-mapping old ones. XOR chunks are cut based on the
|
|
// number of samples they contain with a soft cap in bytes.
|
|
// It is unsafe to call this concurrently with s.iterator(...) without holding the series lock.
|
|
// This should be called only when appending data.
|
|
func (s *memSeries) appendPreprocessor(t int64, e chunkenc.Encoding, o chunkOpts) (c *memChunk, sampleInOrder, chunkCreated bool) {
|
|
// We target chunkenc.MaxBytesPerXORChunk as a hard for the size of an XOR chunk. We must determine whether to cut
|
|
// a new head chunk without knowing the size of the next sample, however, so we assume the next sample will be a
|
|
// maximally-sized sample (19 bytes).
|
|
const maxBytesPerXORChunk = chunkenc.MaxBytesPerXORChunk - 19
|
|
|
|
c = s.headChunks
|
|
|
|
if c == nil {
|
|
if len(s.mmappedChunks) > 0 && s.mmappedChunks[len(s.mmappedChunks)-1].maxTime >= t {
|
|
// Out of order sample. Sample timestamp is already in the mmapped chunks, so ignore it.
|
|
return c, false, false
|
|
}
|
|
// There is no head chunk in this series yet, create the first chunk for the sample.
|
|
c = s.cutNewHeadChunk(t, e, o.chunkRange)
|
|
chunkCreated = true
|
|
}
|
|
|
|
// Out of order sample.
|
|
if c.maxTime >= t {
|
|
return c, false, chunkCreated
|
|
}
|
|
|
|
// Check the chunk size, unless we just created it and if the chunk is too large, cut a new one.
|
|
if !chunkCreated && len(c.chunk.Bytes()) > maxBytesPerXORChunk {
|
|
c = s.cutNewHeadChunk(t, e, o.chunkRange)
|
|
chunkCreated = true
|
|
}
|
|
|
|
if c.chunk.Encoding() != e {
|
|
// The chunk encoding expected by this append is different than the head chunk's
|
|
// encoding. So we cut a new chunk with the expected encoding.
|
|
c = s.cutNewHeadChunk(t, e, o.chunkRange)
|
|
chunkCreated = true
|
|
|
|
}
|
|
|
|
numSamples := c.chunk.NumSamples()
|
|
if numSamples == 0 {
|
|
// It could be the new chunk created after reading the chunk snapshot,
|
|
// hence we fix the minTime of the chunk here.
|
|
c.minTime = t
|
|
s.nextAt = rangeForTimestamp(c.minTime, o.chunkRange)
|
|
}
|
|
|
|
// If we reach 25% of a chunk's desired sample count, predict an end time
|
|
// for this chunk that will try to make samples equally distributed within
|
|
// the remaining chunks in the current chunk range.
|
|
// At latest it must happen at the timestamp set when the chunk was cut.
|
|
if numSamples == o.samplesPerChunk/4 {
|
|
s.nextAt = computeChunkEndTime(c.minTime, c.maxTime, s.nextAt, 4)
|
|
}
|
|
// If numSamples > samplesPerChunk*2 then our previous prediction was invalid,
|
|
// most likely because samples rate has changed and now they are arriving more frequently.
|
|
// Since we assume that the rate is higher, we're being conservative and cutting at 2*samplesPerChunk
|
|
// as we expect more chunks to come.
|
|
// Note that next chunk will have its nextAt recalculated for the new rate.
|
|
if t >= s.nextAt || numSamples >= o.samplesPerChunk*2 {
|
|
c = s.cutNewHeadChunk(t, e, o.chunkRange)
|
|
chunkCreated = true
|
|
}
|
|
|
|
return c, true, chunkCreated
|
|
}
|
|
|
|
// histogramsAppendPreprocessor takes care of cutting new histogram chunks and m-mapping old ones. Histogram chunks are
|
|
// cut based on their size in bytes.
|
|
// It is unsafe to call this concurrently with s.iterator(...) without holding the series lock.
|
|
// This should be called only when appending data.
|
|
func (s *memSeries) histogramsAppendPreprocessor(t int64, e chunkenc.Encoding, o chunkOpts) (c *memChunk, sampleInOrder, chunkCreated bool) {
|
|
c = s.headChunks
|
|
|
|
if c == nil {
|
|
if len(s.mmappedChunks) > 0 && s.mmappedChunks[len(s.mmappedChunks)-1].maxTime >= t {
|
|
// Out of order sample. Sample timestamp is already in the mmapped chunks, so ignore it.
|
|
return c, false, false
|
|
}
|
|
// There is no head chunk in this series yet, create the first chunk for the sample.
|
|
c = s.cutNewHeadChunk(t, e, o.chunkRange)
|
|
chunkCreated = true
|
|
}
|
|
|
|
// Out of order sample.
|
|
if c.maxTime >= t {
|
|
return c, false, chunkCreated
|
|
}
|
|
|
|
if c.chunk.Encoding() != e {
|
|
// The chunk encoding expected by this append is different than the head chunk's
|
|
// encoding. So we cut a new chunk with the expected encoding.
|
|
c = s.cutNewHeadChunk(t, e, o.chunkRange)
|
|
chunkCreated = true
|
|
}
|
|
|
|
numSamples := c.chunk.NumSamples()
|
|
targetBytes := chunkenc.TargetBytesPerHistogramChunk
|
|
numBytes := len(c.chunk.Bytes())
|
|
|
|
if numSamples == 0 {
|
|
// It could be the new chunk created after reading the chunk snapshot,
|
|
// hence we fix the minTime of the chunk here.
|
|
c.minTime = t
|
|
s.nextAt = rangeForTimestamp(c.minTime, o.chunkRange)
|
|
}
|
|
|
|
// Below, we will enforce chunkenc.MinSamplesPerHistogramChunk. There are, however, two cases that supersede it:
|
|
// - The current chunk range is ending before chunkenc.MinSamplesPerHistogramChunk will be satisfied.
|
|
// - s.nextAt was set while loading a chunk snapshot with the intent that a new chunk be cut on the next append.
|
|
var nextChunkRangeStart int64
|
|
if s.histogramChunkHasComputedEndTime {
|
|
nextChunkRangeStart = rangeForTimestamp(c.minTime, o.chunkRange)
|
|
} else {
|
|
// If we haven't yet computed an end time yet, s.nextAt is either set to
|
|
// rangeForTimestamp(c.minTime, o.chunkRange) or was set while loading a chunk snapshot. Either way, we want to
|
|
// skip enforcing chunkenc.MinSamplesPerHistogramChunk.
|
|
nextChunkRangeStart = s.nextAt
|
|
}
|
|
|
|
// If we reach 25% of a chunk's desired maximum size, predict an end time
|
|
// for this chunk that will try to make samples equally distributed within
|
|
// the remaining chunks in the current chunk range.
|
|
// At the latest it must happen at the timestamp set when the chunk was cut.
|
|
if !s.histogramChunkHasComputedEndTime && numBytes >= targetBytes/4 {
|
|
ratioToFull := float64(targetBytes) / float64(numBytes)
|
|
s.nextAt = computeChunkEndTime(c.minTime, c.maxTime, s.nextAt, ratioToFull)
|
|
s.histogramChunkHasComputedEndTime = true
|
|
}
|
|
// If numBytes > targetBytes*2 then our previous prediction was invalid. This could happen if the sample rate has
|
|
// increased or if the bucket/span count has increased.
|
|
// Note that next chunk will have its nextAt recalculated for the new rate.
|
|
if (t >= s.nextAt || numBytes >= targetBytes*2) && (numSamples >= chunkenc.MinSamplesPerHistogramChunk || t >= nextChunkRangeStart) {
|
|
c = s.cutNewHeadChunk(t, e, o.chunkRange)
|
|
chunkCreated = true
|
|
}
|
|
|
|
// The new chunk will also need a new computed end time.
|
|
if chunkCreated {
|
|
s.histogramChunkHasComputedEndTime = false
|
|
}
|
|
|
|
return c, true, chunkCreated
|
|
}
|
|
|
|
// computeChunkEndTime estimates the end timestamp based the beginning of a
|
|
// chunk, its current timestamp and the upper bound up to which we insert data.
|
|
// It assumes that the time range is 1/ratioToFull full.
|
|
// Assuming that the samples will keep arriving at the same rate, it will make the
|
|
// remaining n chunks within this chunk range (before max) equally sized.
|
|
func computeChunkEndTime(start, cur, max int64, ratioToFull float64) int64 {
|
|
n := float64(max-start) / (float64(cur-start+1) * ratioToFull)
|
|
if n <= 1 {
|
|
return max
|
|
}
|
|
return int64(float64(start) + float64(max-start)/math.Floor(n))
|
|
}
|
|
|
|
func (s *memSeries) cutNewHeadChunk(mint int64, e chunkenc.Encoding, chunkRange int64) *memChunk {
|
|
// When cutting a new head chunk we create a new memChunk instance with .prev
|
|
// pointing at the current .headChunks, so it forms a linked list.
|
|
// All but first headChunks list elements will be m-mapped as soon as possible
|
|
// so this is a single element list most of the time.
|
|
s.headChunks = &memChunk{
|
|
minTime: mint,
|
|
maxTime: math.MinInt64,
|
|
prev: s.headChunks,
|
|
}
|
|
|
|
if chunkenc.IsValidEncoding(e) {
|
|
var err error
|
|
s.headChunks.chunk, err = chunkenc.NewEmptyChunk(e)
|
|
if err != nil {
|
|
panic(err) // This should never happen.
|
|
}
|
|
} else {
|
|
s.headChunks.chunk = chunkenc.NewXORChunk()
|
|
}
|
|
|
|
// Set upper bound on when the next chunk must be started. An earlier timestamp
|
|
// may be chosen dynamically at a later point.
|
|
s.nextAt = rangeForTimestamp(mint, chunkRange)
|
|
|
|
app, err := s.headChunks.chunk.Appender()
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
s.app = app
|
|
return s.headChunks
|
|
}
|
|
|
|
// cutNewOOOHeadChunk cuts a new OOO chunk and m-maps the old chunk.
|
|
// The caller must ensure that s.ooo is not nil.
|
|
func (s *memSeries) cutNewOOOHeadChunk(mint int64, chunkDiskMapper *chunks.ChunkDiskMapper) (*oooHeadChunk, chunks.ChunkDiskMapperRef) {
|
|
ref := s.mmapCurrentOOOHeadChunk(chunkDiskMapper)
|
|
|
|
s.ooo.oooHeadChunk = &oooHeadChunk{
|
|
chunk: NewOOOChunk(),
|
|
minTime: mint,
|
|
maxTime: math.MinInt64,
|
|
}
|
|
|
|
return s.ooo.oooHeadChunk, ref
|
|
}
|
|
|
|
func (s *memSeries) mmapCurrentOOOHeadChunk(chunkDiskMapper *chunks.ChunkDiskMapper) chunks.ChunkDiskMapperRef {
|
|
if s.ooo == nil || s.ooo.oooHeadChunk == nil {
|
|
// There is no head chunk, so nothing to m-map here.
|
|
return 0
|
|
}
|
|
xor, _ := s.ooo.oooHeadChunk.chunk.ToXOR() // Encode to XorChunk which is more compact and implements all of the needed functionality.
|
|
chunkRef := chunkDiskMapper.WriteChunk(s.ref, s.ooo.oooHeadChunk.minTime, s.ooo.oooHeadChunk.maxTime, xor, true, handleChunkWriteError)
|
|
s.ooo.oooMmappedChunks = append(s.ooo.oooMmappedChunks, &mmappedChunk{
|
|
ref: chunkRef,
|
|
numSamples: uint16(xor.NumSamples()),
|
|
minTime: s.ooo.oooHeadChunk.minTime,
|
|
maxTime: s.ooo.oooHeadChunk.maxTime,
|
|
})
|
|
s.ooo.oooHeadChunk = nil
|
|
return chunkRef
|
|
}
|
|
|
|
// mmapChunks will m-map all but first chunk on s.headChunks list.
|
|
func (s *memSeries) mmapChunks(chunkDiskMapper *chunks.ChunkDiskMapper) (count int) {
|
|
if s.headChunks == nil || s.headChunks.prev == nil {
|
|
// There is none or only one head chunk, so nothing to m-map here.
|
|
return
|
|
}
|
|
|
|
// Write chunks starting from the oldest one and stop before we get to current s.headChunk.
|
|
// If we have this chain: s.headChunk{t4} -> t3 -> t2 -> t1 -> t0
|
|
// then we need to write chunks t0 to t3, but skip s.headChunks.
|
|
for i := s.headChunks.len() - 1; i > 0; i-- {
|
|
chk := s.headChunks.atOffset(i)
|
|
chunkRef := chunkDiskMapper.WriteChunk(s.ref, chk.minTime, chk.maxTime, chk.chunk, false, handleChunkWriteError)
|
|
s.mmappedChunks = append(s.mmappedChunks, &mmappedChunk{
|
|
ref: chunkRef,
|
|
numSamples: uint16(chk.chunk.NumSamples()),
|
|
minTime: chk.minTime,
|
|
maxTime: chk.maxTime,
|
|
})
|
|
count++
|
|
}
|
|
|
|
// Once we've written out all chunks except s.headChunks we need to unlink these from s.headChunk.
|
|
s.headChunks.prev = nil
|
|
|
|
return count
|
|
}
|
|
|
|
func handleChunkWriteError(err error) {
|
|
if err != nil && !errors.Is(err, chunks.ErrChunkDiskMapperClosed) {
|
|
panic(err)
|
|
}
|
|
}
|
|
|
|
// Rollback removes the samples and exemplars from headAppender and writes any series to WAL.
|
|
func (a *headAppender) Rollback() (err error) {
|
|
if a.closed {
|
|
return ErrAppenderClosed
|
|
}
|
|
defer func() { a.closed = true }()
|
|
defer a.head.metrics.activeAppenders.Dec()
|
|
defer a.head.iso.closeAppend(a.appendID)
|
|
defer a.head.putSeriesBuffer(a.sampleSeries)
|
|
|
|
var series *memSeries
|
|
for i := range a.samples {
|
|
series = a.sampleSeries[i]
|
|
series.Lock()
|
|
series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow)
|
|
series.pendingCommit = false
|
|
series.Unlock()
|
|
}
|
|
for i := range a.histograms {
|
|
series = a.histogramSeries[i]
|
|
series.Lock()
|
|
series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow)
|
|
series.pendingCommit = false
|
|
series.Unlock()
|
|
}
|
|
a.head.putAppendBuffer(a.samples)
|
|
a.head.putExemplarBuffer(a.exemplars)
|
|
a.head.putHistogramBuffer(a.histograms)
|
|
a.head.putFloatHistogramBuffer(a.floatHistograms)
|
|
a.head.putMetadataBuffer(a.metadata)
|
|
a.samples = nil
|
|
a.exemplars = nil
|
|
a.histograms = nil
|
|
a.metadata = nil
|
|
|
|
// Series are created in the head memory regardless of rollback. Thus we have
|
|
// to log them to the WAL in any case.
|
|
return a.log()
|
|
}
|