mirror of https://github.com/prometheus/prometheus
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1841 lines
52 KiB
1841 lines
52 KiB
// Copyright 2016 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 scrape |
|
|
|
import ( |
|
"bufio" |
|
"bytes" |
|
"compress/gzip" |
|
"context" |
|
"fmt" |
|
"io" |
|
"math" |
|
"net/http" |
|
"reflect" |
|
"sort" |
|
"strconv" |
|
"sync" |
|
"time" |
|
"unsafe" |
|
|
|
"github.com/go-kit/log" |
|
"github.com/go-kit/log/level" |
|
"github.com/pkg/errors" |
|
"github.com/prometheus/client_golang/prometheus" |
|
config_util "github.com/prometheus/common/config" |
|
"github.com/prometheus/common/model" |
|
"github.com/prometheus/common/version" |
|
|
|
"github.com/prometheus/prometheus/config" |
|
"github.com/prometheus/prometheus/discovery/targetgroup" |
|
"github.com/prometheus/prometheus/model/exemplar" |
|
"github.com/prometheus/prometheus/model/labels" |
|
"github.com/prometheus/prometheus/model/relabel" |
|
"github.com/prometheus/prometheus/model/textparse" |
|
"github.com/prometheus/prometheus/model/timestamp" |
|
"github.com/prometheus/prometheus/model/value" |
|
"github.com/prometheus/prometheus/storage" |
|
"github.com/prometheus/prometheus/util/pool" |
|
) |
|
|
|
// ScrapeTimestampTolerance is the tolerance for scrape appends timestamps |
|
// alignment, to enable better compression at the TSDB level. |
|
// See https://github.com/prometheus/prometheus/issues/7846 |
|
var ScrapeTimestampTolerance = 2 * time.Millisecond |
|
|
|
// AlignScrapeTimestamps enables the tolerance for scrape appends timestamps described above. |
|
var AlignScrapeTimestamps = true |
|
|
|
var errNameLabelMandatory = fmt.Errorf("missing metric name (%s label)", labels.MetricName) |
|
|
|
var ( |
|
targetIntervalLength = prometheus.NewSummaryVec( |
|
prometheus.SummaryOpts{ |
|
Name: "prometheus_target_interval_length_seconds", |
|
Help: "Actual intervals between scrapes.", |
|
Objectives: map[float64]float64{0.01: 0.001, 0.05: 0.005, 0.5: 0.05, 0.90: 0.01, 0.99: 0.001}, |
|
}, |
|
[]string{"interval"}, |
|
) |
|
targetReloadIntervalLength = prometheus.NewSummaryVec( |
|
prometheus.SummaryOpts{ |
|
Name: "prometheus_target_reload_length_seconds", |
|
Help: "Actual interval to reload the scrape pool with a given configuration.", |
|
Objectives: map[float64]float64{0.01: 0.001, 0.05: 0.005, 0.5: 0.05, 0.90: 0.01, 0.99: 0.001}, |
|
}, |
|
[]string{"interval"}, |
|
) |
|
targetScrapePools = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrape_pools_total", |
|
Help: "Total number of scrape pool creation attempts.", |
|
}, |
|
) |
|
targetScrapePoolsFailed = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrape_pools_failed_total", |
|
Help: "Total number of scrape pool creations that failed.", |
|
}, |
|
) |
|
targetScrapePoolReloads = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrape_pool_reloads_total", |
|
Help: "Total number of scrape pool reloads.", |
|
}, |
|
) |
|
targetScrapePoolReloadsFailed = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrape_pool_reloads_failed_total", |
|
Help: "Total number of failed scrape pool reloads.", |
|
}, |
|
) |
|
targetScrapePoolExceededTargetLimit = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrape_pool_exceeded_target_limit_total", |
|
Help: "Total number of times scrape pools hit the target limit, during sync or config reload.", |
|
}, |
|
) |
|
targetScrapePoolTargetLimit = prometheus.NewGaugeVec( |
|
prometheus.GaugeOpts{ |
|
Name: "prometheus_target_scrape_pool_target_limit", |
|
Help: "Maximum number of targets allowed in this scrape pool.", |
|
}, |
|
[]string{"scrape_job"}, |
|
) |
|
targetScrapePoolTargetsAdded = prometheus.NewGaugeVec( |
|
prometheus.GaugeOpts{ |
|
Name: "prometheus_target_scrape_pool_targets", |
|
Help: "Current number of targets in this scrape pool.", |
|
}, |
|
[]string{"scrape_job"}, |
|
) |
|
targetSyncIntervalLength = prometheus.NewSummaryVec( |
|
prometheus.SummaryOpts{ |
|
Name: "prometheus_target_sync_length_seconds", |
|
Help: "Actual interval to sync the scrape pool.", |
|
Objectives: map[float64]float64{0.01: 0.001, 0.05: 0.005, 0.5: 0.05, 0.90: 0.01, 0.99: 0.001}, |
|
}, |
|
[]string{"scrape_job"}, |
|
) |
|
targetScrapePoolSyncsCounter = prometheus.NewCounterVec( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrape_pool_sync_total", |
|
Help: "Total number of syncs that were executed on a scrape pool.", |
|
}, |
|
[]string{"scrape_job"}, |
|
) |
|
targetScrapeExceededBodySizeLimit = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrapes_exceeded_body_size_limit_total", |
|
Help: "Total number of scrapes that hit the body size limit", |
|
}, |
|
) |
|
targetScrapeSampleLimit = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrapes_exceeded_sample_limit_total", |
|
Help: "Total number of scrapes that hit the sample limit and were rejected.", |
|
}, |
|
) |
|
targetScrapeSampleDuplicate = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrapes_sample_duplicate_timestamp_total", |
|
Help: "Total number of samples rejected due to duplicate timestamps but different values.", |
|
}, |
|
) |
|
targetScrapeSampleOutOfOrder = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrapes_sample_out_of_order_total", |
|
Help: "Total number of samples rejected due to not being out of the expected order.", |
|
}, |
|
) |
|
targetScrapeSampleOutOfBounds = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrapes_sample_out_of_bounds_total", |
|
Help: "Total number of samples rejected due to timestamp falling outside of the time bounds.", |
|
}, |
|
) |
|
targetScrapeCacheFlushForced = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrapes_cache_flush_forced_total", |
|
Help: "How many times a scrape cache was flushed due to getting big while scrapes are failing.", |
|
}, |
|
) |
|
targetScrapeExemplarOutOfOrder = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrapes_exemplar_out_of_order_total", |
|
Help: "Total number of exemplar rejected due to not being out of the expected order.", |
|
}, |
|
) |
|
targetScrapePoolExceededLabelLimits = prometheus.NewCounter( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_scrape_pool_exceeded_label_limits_total", |
|
Help: "Total number of times scrape pools hit the label limits, during sync or config reload.", |
|
}, |
|
) |
|
targetSyncFailed = prometheus.NewCounterVec( |
|
prometheus.CounterOpts{ |
|
Name: "prometheus_target_sync_failed_total", |
|
Help: "Total number of target sync failures.", |
|
}, |
|
[]string{"scrape_job"}, |
|
) |
|
) |
|
|
|
func init() { |
|
prometheus.MustRegister( |
|
targetIntervalLength, |
|
targetReloadIntervalLength, |
|
targetScrapePools, |
|
targetScrapePoolsFailed, |
|
targetScrapePoolReloads, |
|
targetScrapePoolReloadsFailed, |
|
targetSyncIntervalLength, |
|
targetScrapePoolSyncsCounter, |
|
targetScrapeExceededBodySizeLimit, |
|
targetScrapeSampleLimit, |
|
targetScrapeSampleDuplicate, |
|
targetScrapeSampleOutOfOrder, |
|
targetScrapeSampleOutOfBounds, |
|
targetScrapePoolExceededTargetLimit, |
|
targetScrapePoolTargetLimit, |
|
targetScrapePoolTargetsAdded, |
|
targetScrapeCacheFlushForced, |
|
targetMetadataCache, |
|
targetScrapeExemplarOutOfOrder, |
|
targetScrapePoolExceededLabelLimits, |
|
targetSyncFailed, |
|
) |
|
} |
|
|
|
// scrapePool manages scrapes for sets of targets. |
|
type scrapePool struct { |
|
appendable storage.Appendable |
|
logger log.Logger |
|
cancel context.CancelFunc |
|
httpOpts []config_util.HTTPClientOption |
|
|
|
// mtx must not be taken after targetMtx. |
|
mtx sync.Mutex |
|
config *config.ScrapeConfig |
|
client *http.Client |
|
loops map[uint64]loop |
|
|
|
targetMtx sync.Mutex |
|
// activeTargets and loops must always be synchronized to have the same |
|
// set of hashes. |
|
activeTargets map[uint64]*Target |
|
droppedTargets []*Target |
|
|
|
// Constructor for new scrape loops. This is settable for testing convenience. |
|
newLoop func(scrapeLoopOptions) loop |
|
} |
|
|
|
type labelLimits struct { |
|
labelLimit int |
|
labelNameLengthLimit int |
|
labelValueLengthLimit int |
|
} |
|
|
|
type scrapeLoopOptions struct { |
|
target *Target |
|
scraper scraper |
|
sampleLimit int |
|
labelLimits *labelLimits |
|
honorLabels bool |
|
honorTimestamps bool |
|
interval time.Duration |
|
timeout time.Duration |
|
mrc []*relabel.Config |
|
cache *scrapeCache |
|
} |
|
|
|
const maxAheadTime = 10 * time.Minute |
|
|
|
type labelsMutator func(labels.Labels) labels.Labels |
|
|
|
func newScrapePool(cfg *config.ScrapeConfig, app storage.Appendable, jitterSeed uint64, logger log.Logger, reportExtraMetrics, passMetadataInContext bool, httpOpts []config_util.HTTPClientOption) (*scrapePool, error) { |
|
targetScrapePools.Inc() |
|
if logger == nil { |
|
logger = log.NewNopLogger() |
|
} |
|
|
|
client, err := config_util.NewClientFromConfig(cfg.HTTPClientConfig, cfg.JobName, httpOpts...) |
|
if err != nil { |
|
targetScrapePoolsFailed.Inc() |
|
return nil, errors.Wrap(err, "error creating HTTP client") |
|
} |
|
|
|
buffers := pool.New(1e3, 100e6, 3, func(sz int) interface{} { return make([]byte, 0, sz) }) |
|
|
|
ctx, cancel := context.WithCancel(context.Background()) |
|
sp := &scrapePool{ |
|
cancel: cancel, |
|
appendable: app, |
|
config: cfg, |
|
client: client, |
|
activeTargets: map[uint64]*Target{}, |
|
loops: map[uint64]loop{}, |
|
logger: logger, |
|
httpOpts: httpOpts, |
|
} |
|
sp.newLoop = func(opts scrapeLoopOptions) loop { |
|
// Update the targets retrieval function for metadata to a new scrape cache. |
|
cache := opts.cache |
|
if cache == nil { |
|
cache = newScrapeCache() |
|
} |
|
opts.target.SetMetadataStore(cache) |
|
|
|
return newScrapeLoop( |
|
ctx, |
|
opts.scraper, |
|
log.With(logger, "target", opts.target), |
|
buffers, |
|
func(l labels.Labels) labels.Labels { |
|
return mutateSampleLabels(l, opts.target, opts.honorLabels, opts.mrc) |
|
}, |
|
func(l labels.Labels) labels.Labels { return mutateReportSampleLabels(l, opts.target) }, |
|
func(ctx context.Context) storage.Appender { return app.Appender(ctx) }, |
|
cache, |
|
jitterSeed, |
|
opts.honorTimestamps, |
|
opts.sampleLimit, |
|
opts.labelLimits, |
|
opts.interval, |
|
opts.timeout, |
|
reportExtraMetrics, |
|
opts.target, |
|
cache, |
|
passMetadataInContext, |
|
) |
|
} |
|
|
|
return sp, nil |
|
} |
|
|
|
func (sp *scrapePool) ActiveTargets() []*Target { |
|
sp.targetMtx.Lock() |
|
defer sp.targetMtx.Unlock() |
|
|
|
var tActive []*Target |
|
for _, t := range sp.activeTargets { |
|
tActive = append(tActive, t) |
|
} |
|
return tActive |
|
} |
|
|
|
func (sp *scrapePool) DroppedTargets() []*Target { |
|
sp.targetMtx.Lock() |
|
defer sp.targetMtx.Unlock() |
|
return sp.droppedTargets |
|
} |
|
|
|
// stop terminates all scrape loops and returns after they all terminated. |
|
func (sp *scrapePool) stop() { |
|
sp.mtx.Lock() |
|
defer sp.mtx.Unlock() |
|
sp.cancel() |
|
var wg sync.WaitGroup |
|
|
|
sp.targetMtx.Lock() |
|
|
|
for fp, l := range sp.loops { |
|
wg.Add(1) |
|
|
|
go func(l loop) { |
|
l.stop() |
|
wg.Done() |
|
}(l) |
|
|
|
delete(sp.loops, fp) |
|
delete(sp.activeTargets, fp) |
|
} |
|
|
|
sp.targetMtx.Unlock() |
|
|
|
wg.Wait() |
|
sp.client.CloseIdleConnections() |
|
|
|
if sp.config != nil { |
|
targetScrapePoolSyncsCounter.DeleteLabelValues(sp.config.JobName) |
|
targetScrapePoolTargetLimit.DeleteLabelValues(sp.config.JobName) |
|
targetScrapePoolTargetsAdded.DeleteLabelValues(sp.config.JobName) |
|
targetSyncIntervalLength.DeleteLabelValues(sp.config.JobName) |
|
targetSyncFailed.DeleteLabelValues(sp.config.JobName) |
|
} |
|
} |
|
|
|
// reload the scrape pool with the given scrape configuration. The target state is preserved |
|
// but all scrape loops are restarted with the new scrape configuration. |
|
// This method returns after all scrape loops that were stopped have stopped scraping. |
|
func (sp *scrapePool) reload(cfg *config.ScrapeConfig) error { |
|
sp.mtx.Lock() |
|
defer sp.mtx.Unlock() |
|
targetScrapePoolReloads.Inc() |
|
start := time.Now() |
|
|
|
client, err := config_util.NewClientFromConfig(cfg.HTTPClientConfig, cfg.JobName, sp.httpOpts...) |
|
if err != nil { |
|
targetScrapePoolReloadsFailed.Inc() |
|
return errors.Wrap(err, "error creating HTTP client") |
|
} |
|
|
|
reuseCache := reusableCache(sp.config, cfg) |
|
sp.config = cfg |
|
oldClient := sp.client |
|
sp.client = client |
|
|
|
targetScrapePoolTargetLimit.WithLabelValues(sp.config.JobName).Set(float64(sp.config.TargetLimit)) |
|
|
|
var ( |
|
wg sync.WaitGroup |
|
interval = time.Duration(sp.config.ScrapeInterval) |
|
timeout = time.Duration(sp.config.ScrapeTimeout) |
|
bodySizeLimit = int64(sp.config.BodySizeLimit) |
|
sampleLimit = int(sp.config.SampleLimit) |
|
labelLimits = &labelLimits{ |
|
labelLimit: int(sp.config.LabelLimit), |
|
labelNameLengthLimit: int(sp.config.LabelNameLengthLimit), |
|
labelValueLengthLimit: int(sp.config.LabelValueLengthLimit), |
|
} |
|
honorLabels = sp.config.HonorLabels |
|
honorTimestamps = sp.config.HonorTimestamps |
|
mrc = sp.config.MetricRelabelConfigs |
|
) |
|
|
|
sp.targetMtx.Lock() |
|
|
|
forcedErr := sp.refreshTargetLimitErr() |
|
for fp, oldLoop := range sp.loops { |
|
var cache *scrapeCache |
|
if oc := oldLoop.getCache(); reuseCache && oc != nil { |
|
oldLoop.disableEndOfRunStalenessMarkers() |
|
cache = oc |
|
} else { |
|
cache = newScrapeCache() |
|
} |
|
|
|
var ( |
|
t = sp.activeTargets[fp] |
|
s = &targetScraper{Target: t, client: sp.client, timeout: timeout, bodySizeLimit: bodySizeLimit} |
|
newLoop = sp.newLoop(scrapeLoopOptions{ |
|
target: t, |
|
scraper: s, |
|
sampleLimit: sampleLimit, |
|
labelLimits: labelLimits, |
|
honorLabels: honorLabels, |
|
honorTimestamps: honorTimestamps, |
|
mrc: mrc, |
|
cache: cache, |
|
interval: interval, |
|
timeout: timeout, |
|
}) |
|
) |
|
wg.Add(1) |
|
|
|
go func(oldLoop, newLoop loop) { |
|
oldLoop.stop() |
|
wg.Done() |
|
|
|
newLoop.setForcedError(forcedErr) |
|
newLoop.run(nil) |
|
}(oldLoop, newLoop) |
|
|
|
sp.loops[fp] = newLoop |
|
} |
|
|
|
sp.targetMtx.Unlock() |
|
|
|
wg.Wait() |
|
oldClient.CloseIdleConnections() |
|
targetReloadIntervalLength.WithLabelValues(interval.String()).Observe( |
|
time.Since(start).Seconds(), |
|
) |
|
return nil |
|
} |
|
|
|
// Sync converts target groups into actual scrape targets and synchronizes |
|
// the currently running scraper with the resulting set and returns all scraped and dropped targets. |
|
func (sp *scrapePool) Sync(tgs []*targetgroup.Group) { |
|
sp.mtx.Lock() |
|
defer sp.mtx.Unlock() |
|
start := time.Now() |
|
|
|
sp.targetMtx.Lock() |
|
var all []*Target |
|
sp.droppedTargets = []*Target{} |
|
for _, tg := range tgs { |
|
targets, failures := TargetsFromGroup(tg, sp.config) |
|
for _, err := range failures { |
|
level.Error(sp.logger).Log("msg", "Creating target failed", "err", err) |
|
} |
|
targetSyncFailed.WithLabelValues(sp.config.JobName).Add(float64(len(failures))) |
|
for _, t := range targets { |
|
if t.Labels().Len() > 0 { |
|
all = append(all, t) |
|
} else if t.DiscoveredLabels().Len() > 0 { |
|
sp.droppedTargets = append(sp.droppedTargets, t) |
|
} |
|
} |
|
} |
|
sp.targetMtx.Unlock() |
|
sp.sync(all) |
|
|
|
targetSyncIntervalLength.WithLabelValues(sp.config.JobName).Observe( |
|
time.Since(start).Seconds(), |
|
) |
|
targetScrapePoolSyncsCounter.WithLabelValues(sp.config.JobName).Inc() |
|
} |
|
|
|
// sync takes a list of potentially duplicated targets, deduplicates them, starts |
|
// scrape loops for new targets, and stops scrape loops for disappeared targets. |
|
// It returns after all stopped scrape loops terminated. |
|
func (sp *scrapePool) sync(targets []*Target) { |
|
var ( |
|
uniqueLoops = make(map[uint64]loop) |
|
interval = time.Duration(sp.config.ScrapeInterval) |
|
timeout = time.Duration(sp.config.ScrapeTimeout) |
|
bodySizeLimit = int64(sp.config.BodySizeLimit) |
|
sampleLimit = int(sp.config.SampleLimit) |
|
labelLimits = &labelLimits{ |
|
labelLimit: int(sp.config.LabelLimit), |
|
labelNameLengthLimit: int(sp.config.LabelNameLengthLimit), |
|
labelValueLengthLimit: int(sp.config.LabelValueLengthLimit), |
|
} |
|
honorLabels = sp.config.HonorLabels |
|
honorTimestamps = sp.config.HonorTimestamps |
|
mrc = sp.config.MetricRelabelConfigs |
|
) |
|
|
|
sp.targetMtx.Lock() |
|
for _, t := range targets { |
|
hash := t.hash() |
|
|
|
if _, ok := sp.activeTargets[hash]; !ok { |
|
// The scrape interval and timeout labels are set to the config's values initially, |
|
// so whether changed via relabeling or not, they'll exist and hold the correct values |
|
// for every target. |
|
var err error |
|
interval, timeout, err = t.intervalAndTimeout(interval, timeout) |
|
|
|
s := &targetScraper{Target: t, client: sp.client, timeout: timeout, bodySizeLimit: bodySizeLimit} |
|
l := sp.newLoop(scrapeLoopOptions{ |
|
target: t, |
|
scraper: s, |
|
sampleLimit: sampleLimit, |
|
labelLimits: labelLimits, |
|
honorLabels: honorLabels, |
|
honorTimestamps: honorTimestamps, |
|
mrc: mrc, |
|
interval: interval, |
|
timeout: timeout, |
|
}) |
|
if err != nil { |
|
l.setForcedError(err) |
|
} |
|
|
|
sp.activeTargets[hash] = t |
|
sp.loops[hash] = l |
|
|
|
uniqueLoops[hash] = l |
|
} else { |
|
// This might be a duplicated target. |
|
if _, ok := uniqueLoops[hash]; !ok { |
|
uniqueLoops[hash] = nil |
|
} |
|
// Need to keep the most updated labels information |
|
// for displaying it in the Service Discovery web page. |
|
sp.activeTargets[hash].SetDiscoveredLabels(t.DiscoveredLabels()) |
|
} |
|
} |
|
|
|
var wg sync.WaitGroup |
|
|
|
// Stop and remove old targets and scraper loops. |
|
for hash := range sp.activeTargets { |
|
if _, ok := uniqueLoops[hash]; !ok { |
|
wg.Add(1) |
|
go func(l loop) { |
|
l.stop() |
|
wg.Done() |
|
}(sp.loops[hash]) |
|
|
|
delete(sp.loops, hash) |
|
delete(sp.activeTargets, hash) |
|
} |
|
} |
|
|
|
sp.targetMtx.Unlock() |
|
|
|
targetScrapePoolTargetsAdded.WithLabelValues(sp.config.JobName).Set(float64(len(uniqueLoops))) |
|
forcedErr := sp.refreshTargetLimitErr() |
|
for _, l := range sp.loops { |
|
l.setForcedError(forcedErr) |
|
} |
|
for _, l := range uniqueLoops { |
|
if l != nil { |
|
go l.run(nil) |
|
} |
|
} |
|
// Wait for all potentially stopped scrapers to terminate. |
|
// This covers the case of flapping targets. If the server is under high load, a new scraper |
|
// may be active and tries to insert. The old scraper that didn't terminate yet could still |
|
// be inserting a previous sample set. |
|
wg.Wait() |
|
} |
|
|
|
// refreshTargetLimitErr returns an error that can be passed to the scrape loops |
|
// if the number of targets exceeds the configured limit. |
|
func (sp *scrapePool) refreshTargetLimitErr() error { |
|
if sp.config == nil || sp.config.TargetLimit == 0 { |
|
return nil |
|
} |
|
if l := len(sp.activeTargets); l > int(sp.config.TargetLimit) { |
|
targetScrapePoolExceededTargetLimit.Inc() |
|
return fmt.Errorf("target_limit exceeded (number of targets: %d, limit: %d)", l, sp.config.TargetLimit) |
|
} |
|
return nil |
|
} |
|
|
|
func verifyLabelLimits(lset labels.Labels, limits *labelLimits) error { |
|
if limits == nil { |
|
return nil |
|
} |
|
|
|
met := lset.Get(labels.MetricName) |
|
if limits.labelLimit > 0 { |
|
nbLabels := len(lset) |
|
if nbLabels > int(limits.labelLimit) { |
|
return fmt.Errorf("label_limit exceeded (metric: %.50s, number of labels: %d, limit: %d)", met, nbLabels, limits.labelLimit) |
|
} |
|
} |
|
|
|
if limits.labelNameLengthLimit == 0 && limits.labelValueLengthLimit == 0 { |
|
return nil |
|
} |
|
|
|
for _, l := range lset { |
|
if limits.labelNameLengthLimit > 0 { |
|
nameLength := len(l.Name) |
|
if nameLength > int(limits.labelNameLengthLimit) { |
|
return fmt.Errorf("label_name_length_limit exceeded (metric: %.50s, label name: %.50s, length: %d, limit: %d)", met, l.Name, nameLength, limits.labelNameLengthLimit) |
|
} |
|
} |
|
|
|
if limits.labelValueLengthLimit > 0 { |
|
valueLength := len(l.Value) |
|
if valueLength > int(limits.labelValueLengthLimit) { |
|
return fmt.Errorf("label_value_length_limit exceeded (metric: %.50s, label name: %.50s, value: %.50q, length: %d, limit: %d)", met, l.Name, l.Value, valueLength, limits.labelValueLengthLimit) |
|
} |
|
} |
|
} |
|
return nil |
|
} |
|
|
|
func mutateSampleLabels(lset labels.Labels, target *Target, honor bool, rc []*relabel.Config) labels.Labels { |
|
lb := labels.NewBuilder(lset) |
|
targetLabels := target.Labels() |
|
|
|
if honor { |
|
for _, l := range targetLabels { |
|
if !lset.Has(l.Name) { |
|
lb.Set(l.Name, l.Value) |
|
} |
|
} |
|
} else { |
|
var conflictingExposedLabels labels.Labels |
|
for _, l := range targetLabels { |
|
existingValue := lset.Get(l.Name) |
|
if existingValue != "" { |
|
conflictingExposedLabels = append(conflictingExposedLabels, labels.Label{Name: l.Name, Value: existingValue}) |
|
} |
|
// It is now safe to set the target label. |
|
lb.Set(l.Name, l.Value) |
|
} |
|
|
|
if len(conflictingExposedLabels) > 0 { |
|
resolveConflictingExposedLabels(lb, lset, targetLabels, conflictingExposedLabels) |
|
} |
|
} |
|
|
|
res := lb.Labels() |
|
|
|
if len(rc) > 0 { |
|
res = relabel.Process(res, rc...) |
|
} |
|
|
|
return res |
|
} |
|
|
|
func resolveConflictingExposedLabels(lb *labels.Builder, exposedLabels, targetLabels, conflictingExposedLabels labels.Labels) { |
|
sort.SliceStable(conflictingExposedLabels, func(i, j int) bool { |
|
return len(conflictingExposedLabels[i].Name) < len(conflictingExposedLabels[j].Name) |
|
}) |
|
|
|
for i, l := range conflictingExposedLabels { |
|
newName := l.Name |
|
for { |
|
newName = model.ExportedLabelPrefix + newName |
|
if !exposedLabels.Has(newName) && |
|
!targetLabels.Has(newName) && |
|
!conflictingExposedLabels[:i].Has(newName) { |
|
conflictingExposedLabels[i].Name = newName |
|
break |
|
} |
|
} |
|
} |
|
|
|
for _, l := range conflictingExposedLabels { |
|
lb.Set(l.Name, l.Value) |
|
} |
|
} |
|
|
|
func mutateReportSampleLabels(lset labels.Labels, target *Target) labels.Labels { |
|
lb := labels.NewBuilder(lset) |
|
|
|
for _, l := range target.Labels() { |
|
lb.Set(model.ExportedLabelPrefix+l.Name, lset.Get(l.Name)) |
|
lb.Set(l.Name, l.Value) |
|
} |
|
|
|
return lb.Labels() |
|
} |
|
|
|
// appender returns an appender for ingested samples from the target. |
|
func appender(app storage.Appender, limit int) storage.Appender { |
|
app = &timeLimitAppender{ |
|
Appender: app, |
|
maxTime: timestamp.FromTime(time.Now().Add(maxAheadTime)), |
|
} |
|
|
|
// The limit is applied after metrics are potentially dropped via relabeling. |
|
if limit > 0 { |
|
app = &limitAppender{ |
|
Appender: app, |
|
limit: limit, |
|
} |
|
} |
|
return app |
|
} |
|
|
|
// A scraper retrieves samples and accepts a status report at the end. |
|
type scraper interface { |
|
scrape(ctx context.Context, w io.Writer) (string, error) |
|
Report(start time.Time, dur time.Duration, err error) |
|
offset(interval time.Duration, jitterSeed uint64) time.Duration |
|
} |
|
|
|
// targetScraper implements the scraper interface for a target. |
|
type targetScraper struct { |
|
*Target |
|
|
|
client *http.Client |
|
req *http.Request |
|
timeout time.Duration |
|
|
|
gzipr *gzip.Reader |
|
buf *bufio.Reader |
|
|
|
bodySizeLimit int64 |
|
} |
|
|
|
var errBodySizeLimit = errors.New("body size limit exceeded") |
|
|
|
const acceptHeader = `application/openmetrics-text;version=1.0.0,application/openmetrics-text;version=0.0.1;q=0.75,text/plain;version=0.0.4;q=0.5,*/*;q=0.1` |
|
|
|
var UserAgent = fmt.Sprintf("Prometheus/%s", version.Version) |
|
|
|
func (s *targetScraper) scrape(ctx context.Context, w io.Writer) (string, error) { |
|
if s.req == nil { |
|
req, err := http.NewRequest("GET", s.URL().String(), nil) |
|
if err != nil { |
|
return "", err |
|
} |
|
req.Header.Add("Accept", acceptHeader) |
|
req.Header.Add("Accept-Encoding", "gzip") |
|
req.Header.Set("User-Agent", UserAgent) |
|
req.Header.Set("X-Prometheus-Scrape-Timeout-Seconds", strconv.FormatFloat(s.timeout.Seconds(), 'f', -1, 64)) |
|
|
|
s.req = req |
|
} |
|
|
|
resp, err := s.client.Do(s.req.WithContext(ctx)) |
|
if err != nil { |
|
return "", err |
|
} |
|
defer func() { |
|
io.Copy(io.Discard, resp.Body) |
|
resp.Body.Close() |
|
}() |
|
|
|
if resp.StatusCode != http.StatusOK { |
|
return "", errors.Errorf("server returned HTTP status %s", resp.Status) |
|
} |
|
|
|
if s.bodySizeLimit <= 0 { |
|
s.bodySizeLimit = math.MaxInt64 |
|
} |
|
if resp.Header.Get("Content-Encoding") != "gzip" { |
|
n, err := io.Copy(w, io.LimitReader(resp.Body, s.bodySizeLimit)) |
|
if err != nil { |
|
return "", err |
|
} |
|
if n >= s.bodySizeLimit { |
|
targetScrapeExceededBodySizeLimit.Inc() |
|
return "", errBodySizeLimit |
|
} |
|
return resp.Header.Get("Content-Type"), nil |
|
} |
|
|
|
if s.gzipr == nil { |
|
s.buf = bufio.NewReader(resp.Body) |
|
s.gzipr, err = gzip.NewReader(s.buf) |
|
if err != nil { |
|
return "", err |
|
} |
|
} else { |
|
s.buf.Reset(resp.Body) |
|
if err = s.gzipr.Reset(s.buf); err != nil { |
|
return "", err |
|
} |
|
} |
|
|
|
n, err := io.Copy(w, io.LimitReader(s.gzipr, s.bodySizeLimit)) |
|
s.gzipr.Close() |
|
if err != nil { |
|
return "", err |
|
} |
|
if n >= s.bodySizeLimit { |
|
targetScrapeExceededBodySizeLimit.Inc() |
|
return "", errBodySizeLimit |
|
} |
|
return resp.Header.Get("Content-Type"), nil |
|
} |
|
|
|
// A loop can run and be stopped again. It must not be reused after it was stopped. |
|
type loop interface { |
|
run(errc chan<- error) |
|
setForcedError(err error) |
|
stop() |
|
getCache() *scrapeCache |
|
disableEndOfRunStalenessMarkers() |
|
} |
|
|
|
type cacheEntry struct { |
|
ref storage.SeriesRef |
|
lastIter uint64 |
|
hash uint64 |
|
lset labels.Labels |
|
} |
|
|
|
type scrapeLoop struct { |
|
scraper scraper |
|
l log.Logger |
|
cache *scrapeCache |
|
lastScrapeSize int |
|
buffers *pool.Pool |
|
jitterSeed uint64 |
|
honorTimestamps bool |
|
forcedErr error |
|
forcedErrMtx sync.Mutex |
|
sampleLimit int |
|
labelLimits *labelLimits |
|
interval time.Duration |
|
timeout time.Duration |
|
|
|
appender func(ctx context.Context) storage.Appender |
|
sampleMutator labelsMutator |
|
reportSampleMutator labelsMutator |
|
|
|
parentCtx context.Context |
|
appenderCtx context.Context |
|
ctx context.Context |
|
cancel func() |
|
stopped chan struct{} |
|
|
|
disabledEndOfRunStalenessMarkers bool |
|
|
|
reportExtraMetrics bool |
|
} |
|
|
|
// scrapeCache tracks mappings of exposed metric strings to label sets and |
|
// storage references. Additionally, it tracks staleness of series between |
|
// scrapes. |
|
type scrapeCache struct { |
|
iter uint64 // Current scrape iteration. |
|
|
|
// How many series and metadata entries there were at the last success. |
|
successfulCount int |
|
|
|
// Parsed string to an entry with information about the actual label set |
|
// and its storage reference. |
|
series map[string]*cacheEntry |
|
|
|
// Cache of dropped metric strings and their iteration. The iteration must |
|
// be a pointer so we can update it without setting a new entry with an unsafe |
|
// string in addDropped(). |
|
droppedSeries map[string]*uint64 |
|
|
|
// seriesCur and seriesPrev store the labels of series that were seen |
|
// in the current and previous scrape. |
|
// We hold two maps and swap them out to save allocations. |
|
seriesCur map[uint64]labels.Labels |
|
seriesPrev map[uint64]labels.Labels |
|
|
|
metaMtx sync.Mutex |
|
metadata map[string]*metaEntry |
|
} |
|
|
|
// metaEntry holds meta information about a metric. |
|
type metaEntry struct { |
|
lastIter uint64 // Last scrape iteration the entry was observed at. |
|
typ textparse.MetricType |
|
help string |
|
unit string |
|
} |
|
|
|
func (m *metaEntry) size() int { |
|
// The attribute lastIter although part of the struct it is not metadata. |
|
return len(m.help) + len(m.unit) + len(m.typ) |
|
} |
|
|
|
func newScrapeCache() *scrapeCache { |
|
return &scrapeCache{ |
|
series: map[string]*cacheEntry{}, |
|
droppedSeries: map[string]*uint64{}, |
|
seriesCur: map[uint64]labels.Labels{}, |
|
seriesPrev: map[uint64]labels.Labels{}, |
|
metadata: map[string]*metaEntry{}, |
|
} |
|
} |
|
|
|
func (c *scrapeCache) iterDone(flushCache bool) { |
|
c.metaMtx.Lock() |
|
count := len(c.series) + len(c.droppedSeries) + len(c.metadata) |
|
c.metaMtx.Unlock() |
|
|
|
if flushCache { |
|
c.successfulCount = count |
|
} else if count > c.successfulCount*2+1000 { |
|
// If a target had varying labels in scrapes that ultimately failed, |
|
// the caches would grow indefinitely. Force a flush when this happens. |
|
// We use the heuristic that this is a doubling of the cache size |
|
// since the last scrape, and allow an additional 1000 in case |
|
// initial scrapes all fail. |
|
flushCache = true |
|
targetScrapeCacheFlushForced.Inc() |
|
} |
|
|
|
if flushCache { |
|
// All caches may grow over time through series churn |
|
// or multiple string representations of the same metric. Clean up entries |
|
// that haven't appeared in the last scrape. |
|
for s, e := range c.series { |
|
if c.iter != e.lastIter { |
|
delete(c.series, s) |
|
} |
|
} |
|
for s, iter := range c.droppedSeries { |
|
if c.iter != *iter { |
|
delete(c.droppedSeries, s) |
|
} |
|
} |
|
c.metaMtx.Lock() |
|
for m, e := range c.metadata { |
|
// Keep metadata around for 10 scrapes after its metric disappeared. |
|
if c.iter-e.lastIter > 10 { |
|
delete(c.metadata, m) |
|
} |
|
} |
|
c.metaMtx.Unlock() |
|
|
|
c.iter++ |
|
} |
|
|
|
// Swap current and previous series. |
|
c.seriesPrev, c.seriesCur = c.seriesCur, c.seriesPrev |
|
|
|
// We have to delete every single key in the map. |
|
for k := range c.seriesCur { |
|
delete(c.seriesCur, k) |
|
} |
|
} |
|
|
|
func (c *scrapeCache) get(met string) (*cacheEntry, bool) { |
|
e, ok := c.series[met] |
|
if !ok { |
|
return nil, false |
|
} |
|
e.lastIter = c.iter |
|
return e, true |
|
} |
|
|
|
func (c *scrapeCache) addRef(met string, ref storage.SeriesRef, lset labels.Labels, hash uint64) { |
|
if ref == 0 { |
|
return |
|
} |
|
c.series[met] = &cacheEntry{ref: ref, lastIter: c.iter, lset: lset, hash: hash} |
|
} |
|
|
|
func (c *scrapeCache) addDropped(met string) { |
|
iter := c.iter |
|
c.droppedSeries[met] = &iter |
|
} |
|
|
|
func (c *scrapeCache) getDropped(met string) bool { |
|
iterp, ok := c.droppedSeries[met] |
|
if ok { |
|
*iterp = c.iter |
|
} |
|
return ok |
|
} |
|
|
|
func (c *scrapeCache) trackStaleness(hash uint64, lset labels.Labels) { |
|
c.seriesCur[hash] = lset |
|
} |
|
|
|
func (c *scrapeCache) forEachStale(f func(labels.Labels) bool) { |
|
for h, lset := range c.seriesPrev { |
|
if _, ok := c.seriesCur[h]; !ok { |
|
if !f(lset) { |
|
break |
|
} |
|
} |
|
} |
|
} |
|
|
|
func (c *scrapeCache) setType(metric []byte, t textparse.MetricType) { |
|
c.metaMtx.Lock() |
|
|
|
e, ok := c.metadata[yoloString(metric)] |
|
if !ok { |
|
e = &metaEntry{typ: textparse.MetricTypeUnknown} |
|
c.metadata[string(metric)] = e |
|
} |
|
e.typ = t |
|
e.lastIter = c.iter |
|
|
|
c.metaMtx.Unlock() |
|
} |
|
|
|
func (c *scrapeCache) setHelp(metric, help []byte) { |
|
c.metaMtx.Lock() |
|
|
|
e, ok := c.metadata[yoloString(metric)] |
|
if !ok { |
|
e = &metaEntry{typ: textparse.MetricTypeUnknown} |
|
c.metadata[string(metric)] = e |
|
} |
|
if e.help != yoloString(help) { |
|
e.help = string(help) |
|
} |
|
e.lastIter = c.iter |
|
|
|
c.metaMtx.Unlock() |
|
} |
|
|
|
func (c *scrapeCache) setUnit(metric, unit []byte) { |
|
c.metaMtx.Lock() |
|
|
|
e, ok := c.metadata[yoloString(metric)] |
|
if !ok { |
|
e = &metaEntry{typ: textparse.MetricTypeUnknown} |
|
c.metadata[string(metric)] = e |
|
} |
|
if e.unit != yoloString(unit) { |
|
e.unit = string(unit) |
|
} |
|
e.lastIter = c.iter |
|
|
|
c.metaMtx.Unlock() |
|
} |
|
|
|
func (c *scrapeCache) GetMetadata(metric string) (MetricMetadata, bool) { |
|
c.metaMtx.Lock() |
|
defer c.metaMtx.Unlock() |
|
|
|
m, ok := c.metadata[metric] |
|
if !ok { |
|
return MetricMetadata{}, false |
|
} |
|
return MetricMetadata{ |
|
Metric: metric, |
|
Type: m.typ, |
|
Help: m.help, |
|
Unit: m.unit, |
|
}, true |
|
} |
|
|
|
func (c *scrapeCache) ListMetadata() []MetricMetadata { |
|
c.metaMtx.Lock() |
|
defer c.metaMtx.Unlock() |
|
|
|
res := make([]MetricMetadata, 0, len(c.metadata)) |
|
|
|
for m, e := range c.metadata { |
|
res = append(res, MetricMetadata{ |
|
Metric: m, |
|
Type: e.typ, |
|
Help: e.help, |
|
Unit: e.unit, |
|
}) |
|
} |
|
return res |
|
} |
|
|
|
// MetadataSize returns the size of the metadata cache. |
|
func (c *scrapeCache) SizeMetadata() (s int) { |
|
c.metaMtx.Lock() |
|
defer c.metaMtx.Unlock() |
|
for _, e := range c.metadata { |
|
s += e.size() |
|
} |
|
|
|
return s |
|
} |
|
|
|
// MetadataLen returns the number of metadata entries in the cache. |
|
func (c *scrapeCache) LengthMetadata() int { |
|
c.metaMtx.Lock() |
|
defer c.metaMtx.Unlock() |
|
|
|
return len(c.metadata) |
|
} |
|
|
|
func newScrapeLoop(ctx context.Context, |
|
sc scraper, |
|
l log.Logger, |
|
buffers *pool.Pool, |
|
sampleMutator labelsMutator, |
|
reportSampleMutator labelsMutator, |
|
appender func(ctx context.Context) storage.Appender, |
|
cache *scrapeCache, |
|
jitterSeed uint64, |
|
honorTimestamps bool, |
|
sampleLimit int, |
|
labelLimits *labelLimits, |
|
interval time.Duration, |
|
timeout time.Duration, |
|
reportExtraMetrics bool, |
|
target *Target, |
|
metricMetadataStore MetricMetadataStore, |
|
passMetadataInContext bool, |
|
) *scrapeLoop { |
|
if l == nil { |
|
l = log.NewNopLogger() |
|
} |
|
if buffers == nil { |
|
buffers = pool.New(1e3, 1e6, 3, func(sz int) interface{} { return make([]byte, 0, sz) }) |
|
} |
|
if cache == nil { |
|
cache = newScrapeCache() |
|
} |
|
|
|
appenderCtx := ctx |
|
|
|
if passMetadataInContext { |
|
// Store the cache and target in the context. This is then used by downstream OTel Collector |
|
// to lookup the metadata required to process the samples. Not used by Prometheus itself. |
|
// TODO(gouthamve) We're using a dedicated context because using the parentCtx caused a memory |
|
// leak. We should ideally fix the main leak. See: https://github.com/prometheus/prometheus/pull/10590 |
|
appenderCtx = ContextWithMetricMetadataStore(appenderCtx, cache) |
|
appenderCtx = ContextWithTarget(appenderCtx, target) |
|
} |
|
|
|
sl := &scrapeLoop{ |
|
scraper: sc, |
|
buffers: buffers, |
|
cache: cache, |
|
appender: appender, |
|
sampleMutator: sampleMutator, |
|
reportSampleMutator: reportSampleMutator, |
|
stopped: make(chan struct{}), |
|
jitterSeed: jitterSeed, |
|
l: l, |
|
parentCtx: ctx, |
|
appenderCtx: appenderCtx, |
|
honorTimestamps: honorTimestamps, |
|
sampleLimit: sampleLimit, |
|
labelLimits: labelLimits, |
|
interval: interval, |
|
timeout: timeout, |
|
reportExtraMetrics: reportExtraMetrics, |
|
} |
|
sl.ctx, sl.cancel = context.WithCancel(ctx) |
|
|
|
return sl |
|
} |
|
|
|
func (sl *scrapeLoop) run(errc chan<- error) { |
|
select { |
|
case <-time.After(sl.scraper.offset(sl.interval, sl.jitterSeed)): |
|
// Continue after a scraping offset. |
|
case <-sl.ctx.Done(): |
|
close(sl.stopped) |
|
return |
|
} |
|
|
|
var last time.Time |
|
|
|
alignedScrapeTime := time.Now().Round(0) |
|
ticker := time.NewTicker(sl.interval) |
|
defer ticker.Stop() |
|
|
|
mainLoop: |
|
for { |
|
select { |
|
case <-sl.parentCtx.Done(): |
|
close(sl.stopped) |
|
return |
|
case <-sl.ctx.Done(): |
|
break mainLoop |
|
default: |
|
} |
|
|
|
// Temporary workaround for a jitter in go timers that causes disk space |
|
// increase in TSDB. |
|
// See https://github.com/prometheus/prometheus/issues/7846 |
|
// Calling Round ensures the time used is the wall clock, as otherwise .Sub |
|
// and .Add on time.Time behave differently (see time package docs). |
|
scrapeTime := time.Now().Round(0) |
|
if AlignScrapeTimestamps && sl.interval > 100*ScrapeTimestampTolerance { |
|
// For some reason, a tick might have been skipped, in which case we |
|
// would call alignedScrapeTime.Add(interval) multiple times. |
|
for scrapeTime.Sub(alignedScrapeTime) >= sl.interval { |
|
alignedScrapeTime = alignedScrapeTime.Add(sl.interval) |
|
} |
|
// Align the scrape time if we are in the tolerance boundaries. |
|
if scrapeTime.Sub(alignedScrapeTime) <= ScrapeTimestampTolerance { |
|
scrapeTime = alignedScrapeTime |
|
} |
|
} |
|
|
|
last = sl.scrapeAndReport(last, scrapeTime, errc) |
|
|
|
select { |
|
case <-sl.parentCtx.Done(): |
|
close(sl.stopped) |
|
return |
|
case <-sl.ctx.Done(): |
|
break mainLoop |
|
case <-ticker.C: |
|
} |
|
} |
|
|
|
close(sl.stopped) |
|
|
|
if !sl.disabledEndOfRunStalenessMarkers { |
|
sl.endOfRunStaleness(last, ticker, sl.interval) |
|
} |
|
} |
|
|
|
// scrapeAndReport performs a scrape and then appends the result to the storage |
|
// together with reporting metrics, by using as few appenders as possible. |
|
// In the happy scenario, a single appender is used. |
|
// This function uses sl.appenderCtx instead of sl.ctx on purpose. A scrape should |
|
// only be cancelled on shutdown, not on reloads. |
|
func (sl *scrapeLoop) scrapeAndReport(last, appendTime time.Time, errc chan<- error) time.Time { |
|
start := time.Now() |
|
|
|
// Only record after the first scrape. |
|
if !last.IsZero() { |
|
targetIntervalLength.WithLabelValues(sl.interval.String()).Observe( |
|
time.Since(last).Seconds(), |
|
) |
|
} |
|
|
|
b := sl.buffers.Get(sl.lastScrapeSize).([]byte) |
|
defer sl.buffers.Put(b) |
|
buf := bytes.NewBuffer(b) |
|
|
|
var total, added, seriesAdded, bytes int |
|
var err, appErr, scrapeErr error |
|
|
|
app := sl.appender(sl.appenderCtx) |
|
defer func() { |
|
if err != nil { |
|
app.Rollback() |
|
return |
|
} |
|
err = app.Commit() |
|
if err != nil { |
|
level.Error(sl.l).Log("msg", "Scrape commit failed", "err", err) |
|
} |
|
}() |
|
|
|
defer func() { |
|
if err = sl.report(app, appendTime, time.Since(start), total, added, seriesAdded, bytes, scrapeErr); err != nil { |
|
level.Warn(sl.l).Log("msg", "Appending scrape report failed", "err", err) |
|
} |
|
}() |
|
|
|
if forcedErr := sl.getForcedError(); forcedErr != nil { |
|
scrapeErr = forcedErr |
|
// Add stale markers. |
|
if _, _, _, err := sl.append(app, []byte{}, "", appendTime); err != nil { |
|
app.Rollback() |
|
app = sl.appender(sl.appenderCtx) |
|
level.Warn(sl.l).Log("msg", "Append failed", "err", err) |
|
} |
|
if errc != nil { |
|
errc <- forcedErr |
|
} |
|
|
|
return start |
|
} |
|
|
|
var contentType string |
|
scrapeCtx, cancel := context.WithTimeout(sl.parentCtx, sl.timeout) |
|
contentType, scrapeErr = sl.scraper.scrape(scrapeCtx, buf) |
|
cancel() |
|
|
|
if scrapeErr == nil { |
|
b = buf.Bytes() |
|
// NOTE: There were issues with misbehaving clients in the past |
|
// that occasionally returned empty results. We don't want those |
|
// to falsely reset our buffer size. |
|
if len(b) > 0 { |
|
sl.lastScrapeSize = len(b) |
|
} |
|
bytes = len(b) |
|
} else { |
|
level.Debug(sl.l).Log("msg", "Scrape failed", "err", scrapeErr) |
|
if errc != nil { |
|
errc <- scrapeErr |
|
} |
|
if errors.Is(scrapeErr, errBodySizeLimit) { |
|
bytes = -1 |
|
} |
|
} |
|
|
|
// A failed scrape is the same as an empty scrape, |
|
// we still call sl.append to trigger stale markers. |
|
total, added, seriesAdded, appErr = sl.append(app, b, contentType, appendTime) |
|
if appErr != nil { |
|
app.Rollback() |
|
app = sl.appender(sl.appenderCtx) |
|
level.Debug(sl.l).Log("msg", "Append failed", "err", appErr) |
|
// The append failed, probably due to a parse error or sample limit. |
|
// Call sl.append again with an empty scrape to trigger stale markers. |
|
if _, _, _, err := sl.append(app, []byte{}, "", appendTime); err != nil { |
|
app.Rollback() |
|
app = sl.appender(sl.appenderCtx) |
|
level.Warn(sl.l).Log("msg", "Append failed", "err", err) |
|
} |
|
} |
|
|
|
if scrapeErr == nil { |
|
scrapeErr = appErr |
|
} |
|
|
|
return start |
|
} |
|
|
|
func (sl *scrapeLoop) setForcedError(err error) { |
|
sl.forcedErrMtx.Lock() |
|
defer sl.forcedErrMtx.Unlock() |
|
sl.forcedErr = err |
|
} |
|
|
|
func (sl *scrapeLoop) getForcedError() error { |
|
sl.forcedErrMtx.Lock() |
|
defer sl.forcedErrMtx.Unlock() |
|
return sl.forcedErr |
|
} |
|
|
|
func (sl *scrapeLoop) endOfRunStaleness(last time.Time, ticker *time.Ticker, interval time.Duration) { |
|
// Scraping has stopped. We want to write stale markers but |
|
// the target may be recreated, so we wait just over 2 scrape intervals |
|
// before creating them. |
|
// If the context is canceled, we presume the server is shutting down |
|
// and will restart where is was. We do not attempt to write stale markers |
|
// in this case. |
|
|
|
if last.IsZero() { |
|
// There never was a scrape, so there will be no stale markers. |
|
return |
|
} |
|
|
|
// Wait for when the next scrape would have been, record its timestamp. |
|
var staleTime time.Time |
|
select { |
|
case <-sl.parentCtx.Done(): |
|
return |
|
case <-ticker.C: |
|
staleTime = time.Now() |
|
} |
|
|
|
// Wait for when the next scrape would have been, if the target was recreated |
|
// samples should have been ingested by now. |
|
select { |
|
case <-sl.parentCtx.Done(): |
|
return |
|
case <-ticker.C: |
|
} |
|
|
|
// Wait for an extra 10% of the interval, just to be safe. |
|
select { |
|
case <-sl.parentCtx.Done(): |
|
return |
|
case <-time.After(interval / 10): |
|
} |
|
|
|
// Call sl.append again with an empty scrape to trigger stale markers. |
|
// If the target has since been recreated and scraped, the |
|
// stale markers will be out of order and ignored. |
|
// sl.context would have been cancelled, hence using sl.appenderCtx. |
|
app := sl.appender(sl.appenderCtx) |
|
var err error |
|
defer func() { |
|
if err != nil { |
|
app.Rollback() |
|
return |
|
} |
|
err = app.Commit() |
|
if err != nil { |
|
level.Warn(sl.l).Log("msg", "Stale commit failed", "err", err) |
|
} |
|
}() |
|
if _, _, _, err = sl.append(app, []byte{}, "", staleTime); err != nil { |
|
app.Rollback() |
|
app = sl.appender(sl.appenderCtx) |
|
level.Warn(sl.l).Log("msg", "Stale append failed", "err", err) |
|
} |
|
if err = sl.reportStale(app, staleTime); err != nil { |
|
level.Warn(sl.l).Log("msg", "Stale report failed", "err", err) |
|
} |
|
} |
|
|
|
// Stop the scraping. May still write data and stale markers after it has |
|
// returned. Cancel the context to stop all writes. |
|
func (sl *scrapeLoop) stop() { |
|
sl.cancel() |
|
<-sl.stopped |
|
} |
|
|
|
func (sl *scrapeLoop) disableEndOfRunStalenessMarkers() { |
|
sl.disabledEndOfRunStalenessMarkers = true |
|
} |
|
|
|
func (sl *scrapeLoop) getCache() *scrapeCache { |
|
return sl.cache |
|
} |
|
|
|
type appendErrors struct { |
|
numOutOfOrder int |
|
numDuplicates int |
|
numOutOfBounds int |
|
numExemplarOutOfOrder int |
|
} |
|
|
|
func (sl *scrapeLoop) append(app storage.Appender, b []byte, contentType string, ts time.Time) (total, added, seriesAdded int, err error) { |
|
p, err := textparse.New(b, contentType) |
|
if err != nil { |
|
level.Debug(sl.l).Log( |
|
"msg", "Invalid content type on scrape, using prometheus parser as fallback.", |
|
"content_type", contentType, |
|
"err", err, |
|
) |
|
} |
|
|
|
var ( |
|
defTime = timestamp.FromTime(ts) |
|
appErrs = appendErrors{} |
|
sampleLimitErr error |
|
e exemplar.Exemplar // escapes to heap so hoisted out of loop |
|
) |
|
|
|
// Take an appender with limits. |
|
app = appender(app, sl.sampleLimit) |
|
|
|
defer func() { |
|
if err != nil { |
|
return |
|
} |
|
// Only perform cache cleaning if the scrape was not empty. |
|
// An empty scrape (usually) is used to indicate a failed scrape. |
|
sl.cache.iterDone(len(b) > 0) |
|
}() |
|
|
|
loop: |
|
for { |
|
var ( |
|
et textparse.Entry |
|
sampleAdded bool |
|
) |
|
if et, err = p.Next(); err != nil { |
|
if err == io.EOF { |
|
err = nil |
|
} |
|
break |
|
} |
|
switch et { |
|
case textparse.EntryType: |
|
sl.cache.setType(p.Type()) |
|
continue |
|
case textparse.EntryHelp: |
|
sl.cache.setHelp(p.Help()) |
|
continue |
|
case textparse.EntryUnit: |
|
sl.cache.setUnit(p.Unit()) |
|
continue |
|
case textparse.EntryComment: |
|
continue |
|
default: |
|
} |
|
total++ |
|
|
|
t := defTime |
|
met, tp, v := p.Series() |
|
if !sl.honorTimestamps { |
|
tp = nil |
|
} |
|
if tp != nil { |
|
t = *tp |
|
} |
|
|
|
if sl.cache.getDropped(yoloString(met)) { |
|
continue |
|
} |
|
ce, ok := sl.cache.get(yoloString(met)) |
|
var ( |
|
ref storage.SeriesRef |
|
lset labels.Labels |
|
mets string |
|
hash uint64 |
|
) |
|
|
|
if ok { |
|
ref = ce.ref |
|
lset = ce.lset |
|
} else { |
|
mets = p.Metric(&lset) |
|
hash = lset.Hash() |
|
|
|
// Hash label set as it is seen local to the target. Then add target labels |
|
// and relabeling and store the final label set. |
|
lset = sl.sampleMutator(lset) |
|
|
|
// The label set may be set to nil to indicate dropping. |
|
if lset == nil { |
|
sl.cache.addDropped(mets) |
|
continue |
|
} |
|
|
|
if !lset.Has(labels.MetricName) { |
|
err = errNameLabelMandatory |
|
break loop |
|
} |
|
|
|
// If any label limits is exceeded the scrape should fail. |
|
if err = verifyLabelLimits(lset, sl.labelLimits); err != nil { |
|
targetScrapePoolExceededLabelLimits.Inc() |
|
break loop |
|
} |
|
} |
|
|
|
ref, err = app.Append(ref, lset, t, v) |
|
sampleAdded, err = sl.checkAddError(ce, met, tp, err, &sampleLimitErr, &appErrs) |
|
if err != nil { |
|
if err != storage.ErrNotFound { |
|
level.Debug(sl.l).Log("msg", "Unexpected error", "series", string(met), "err", err) |
|
} |
|
break loop |
|
} |
|
|
|
if !ok { |
|
if tp == nil { |
|
// Bypass staleness logic if there is an explicit timestamp. |
|
sl.cache.trackStaleness(hash, lset) |
|
} |
|
sl.cache.addRef(mets, ref, lset, hash) |
|
if sampleAdded && sampleLimitErr == nil { |
|
seriesAdded++ |
|
} |
|
} |
|
|
|
// Increment added even if there's an error so we correctly report the |
|
// number of samples remaining after relabeling. |
|
added++ |
|
|
|
if hasExemplar := p.Exemplar(&e); hasExemplar { |
|
if !e.HasTs { |
|
e.Ts = t |
|
} |
|
_, exemplarErr := app.AppendExemplar(ref, lset, e) |
|
exemplarErr = sl.checkAddExemplarError(exemplarErr, e, &appErrs) |
|
if exemplarErr != nil { |
|
// Since exemplar storage is still experimental, we don't fail the scrape on ingestion errors. |
|
level.Debug(sl.l).Log("msg", "Error while adding exemplar in AddExemplar", "exemplar", fmt.Sprintf("%+v", e), "err", exemplarErr) |
|
} |
|
e = exemplar.Exemplar{} // reset for next time round loop |
|
} |
|
|
|
} |
|
if sampleLimitErr != nil { |
|
if err == nil { |
|
err = sampleLimitErr |
|
} |
|
// We only want to increment this once per scrape, so this is Inc'd outside the loop. |
|
targetScrapeSampleLimit.Inc() |
|
} |
|
if appErrs.numOutOfOrder > 0 { |
|
level.Warn(sl.l).Log("msg", "Error on ingesting out-of-order samples", "num_dropped", appErrs.numOutOfOrder) |
|
} |
|
if appErrs.numDuplicates > 0 { |
|
level.Warn(sl.l).Log("msg", "Error on ingesting samples with different value but same timestamp", "num_dropped", appErrs.numDuplicates) |
|
} |
|
if appErrs.numOutOfBounds > 0 { |
|
level.Warn(sl.l).Log("msg", "Error on ingesting samples that are too old or are too far into the future", "num_dropped", appErrs.numOutOfBounds) |
|
} |
|
if appErrs.numExemplarOutOfOrder > 0 { |
|
level.Warn(sl.l).Log("msg", "Error on ingesting out-of-order exemplars", "num_dropped", appErrs.numExemplarOutOfOrder) |
|
} |
|
if err == nil { |
|
sl.cache.forEachStale(func(lset labels.Labels) bool { |
|
// Series no longer exposed, mark it stale. |
|
_, err = app.Append(0, lset, defTime, math.Float64frombits(value.StaleNaN)) |
|
switch errors.Cause(err) { |
|
case storage.ErrOutOfOrderSample, storage.ErrDuplicateSampleForTimestamp: |
|
// Do not count these in logging, as this is expected if a target |
|
// goes away and comes back again with a new scrape loop. |
|
err = nil |
|
} |
|
return err == nil |
|
}) |
|
} |
|
return |
|
} |
|
|
|
func yoloString(b []byte) string { |
|
return *((*string)(unsafe.Pointer(&b))) |
|
} |
|
|
|
// Adds samples to the appender, checking the error, and then returns the # of samples added, |
|
// whether the caller should continue to process more samples, and any sample limit errors. |
|
func (sl *scrapeLoop) checkAddError(ce *cacheEntry, met []byte, tp *int64, err error, sampleLimitErr *error, appErrs *appendErrors) (bool, error) { |
|
switch errors.Cause(err) { |
|
case nil: |
|
if tp == nil && ce != nil { |
|
sl.cache.trackStaleness(ce.hash, ce.lset) |
|
} |
|
return true, nil |
|
case storage.ErrNotFound: |
|
return false, storage.ErrNotFound |
|
case storage.ErrOutOfOrderSample: |
|
appErrs.numOutOfOrder++ |
|
level.Debug(sl.l).Log("msg", "Out of order sample", "series", string(met)) |
|
targetScrapeSampleOutOfOrder.Inc() |
|
return false, nil |
|
case storage.ErrDuplicateSampleForTimestamp: |
|
appErrs.numDuplicates++ |
|
level.Debug(sl.l).Log("msg", "Duplicate sample for timestamp", "series", string(met)) |
|
targetScrapeSampleDuplicate.Inc() |
|
return false, nil |
|
case storage.ErrOutOfBounds: |
|
appErrs.numOutOfBounds++ |
|
level.Debug(sl.l).Log("msg", "Out of bounds metric", "series", string(met)) |
|
targetScrapeSampleOutOfBounds.Inc() |
|
return false, nil |
|
case errSampleLimit: |
|
// Keep on parsing output if we hit the limit, so we report the correct |
|
// total number of samples scraped. |
|
*sampleLimitErr = err |
|
return false, nil |
|
default: |
|
return false, err |
|
} |
|
} |
|
|
|
func (sl *scrapeLoop) checkAddExemplarError(err error, e exemplar.Exemplar, appErrs *appendErrors) error { |
|
switch errors.Cause(err) { |
|
case storage.ErrNotFound: |
|
return storage.ErrNotFound |
|
case storage.ErrOutOfOrderExemplar: |
|
appErrs.numExemplarOutOfOrder++ |
|
level.Debug(sl.l).Log("msg", "Out of order exemplar", "exemplar", fmt.Sprintf("%+v", e)) |
|
targetScrapeExemplarOutOfOrder.Inc() |
|
return nil |
|
default: |
|
return err |
|
} |
|
} |
|
|
|
// The constants are suffixed with the invalid \xff unicode rune to avoid collisions |
|
// with scraped metrics in the cache. |
|
const ( |
|
scrapeHealthMetricName = "up" + "\xff" |
|
scrapeDurationMetricName = "scrape_duration_seconds" + "\xff" |
|
scrapeSamplesMetricName = "scrape_samples_scraped" + "\xff" |
|
samplesPostRelabelMetricName = "scrape_samples_post_metric_relabeling" + "\xff" |
|
scrapeSeriesAddedMetricName = "scrape_series_added" + "\xff" |
|
scrapeTimeoutMetricName = "scrape_timeout_seconds" + "\xff" |
|
scrapeSampleLimitMetricName = "scrape_sample_limit" + "\xff" |
|
scrapeBodySizeBytesMetricName = "scrape_body_size_bytes" + "\xff" |
|
) |
|
|
|
func (sl *scrapeLoop) report(app storage.Appender, start time.Time, duration time.Duration, scraped, added, seriesAdded, bytes int, scrapeErr error) (err error) { |
|
sl.scraper.Report(start, duration, scrapeErr) |
|
|
|
ts := timestamp.FromTime(start) |
|
|
|
var health float64 |
|
if scrapeErr == nil { |
|
health = 1 |
|
} |
|
|
|
if err = sl.addReportSample(app, scrapeHealthMetricName, ts, health); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeDurationMetricName, ts, duration.Seconds()); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeSamplesMetricName, ts, float64(scraped)); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, samplesPostRelabelMetricName, ts, float64(added)); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeSeriesAddedMetricName, ts, float64(seriesAdded)); err != nil { |
|
return |
|
} |
|
if sl.reportExtraMetrics { |
|
if err = sl.addReportSample(app, scrapeTimeoutMetricName, ts, sl.timeout.Seconds()); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeSampleLimitMetricName, ts, float64(sl.sampleLimit)); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeBodySizeBytesMetricName, ts, float64(bytes)); err != nil { |
|
return |
|
} |
|
} |
|
return |
|
} |
|
|
|
func (sl *scrapeLoop) reportStale(app storage.Appender, start time.Time) (err error) { |
|
ts := timestamp.FromTime(start) |
|
|
|
stale := math.Float64frombits(value.StaleNaN) |
|
|
|
if err = sl.addReportSample(app, scrapeHealthMetricName, ts, stale); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeDurationMetricName, ts, stale); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeSamplesMetricName, ts, stale); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, samplesPostRelabelMetricName, ts, stale); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeSeriesAddedMetricName, ts, stale); err != nil { |
|
return |
|
} |
|
if sl.reportExtraMetrics { |
|
if err = sl.addReportSample(app, scrapeTimeoutMetricName, ts, stale); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeSampleLimitMetricName, ts, stale); err != nil { |
|
return |
|
} |
|
if err = sl.addReportSample(app, scrapeBodySizeBytesMetricName, ts, stale); err != nil { |
|
return |
|
} |
|
} |
|
return |
|
} |
|
|
|
func (sl *scrapeLoop) addReportSample(app storage.Appender, s string, t int64, v float64) error { |
|
ce, ok := sl.cache.get(s) |
|
var ref storage.SeriesRef |
|
var lset labels.Labels |
|
if ok { |
|
ref = ce.ref |
|
lset = ce.lset |
|
} else { |
|
lset = labels.Labels{ |
|
// The constants are suffixed with the invalid \xff unicode rune to avoid collisions |
|
// with scraped metrics in the cache. |
|
// We have to drop it when building the actual metric. |
|
labels.Label{Name: labels.MetricName, Value: s[:len(s)-1]}, |
|
} |
|
lset = sl.reportSampleMutator(lset) |
|
} |
|
|
|
ref, err := app.Append(ref, lset, t, v) |
|
switch errors.Cause(err) { |
|
case nil: |
|
if !ok { |
|
sl.cache.addRef(s, ref, lset, lset.Hash()) |
|
} |
|
return nil |
|
case storage.ErrOutOfOrderSample, storage.ErrDuplicateSampleForTimestamp: |
|
// Do not log here, as this is expected if a target goes away and comes back |
|
// again with a new scrape loop. |
|
return nil |
|
default: |
|
return err |
|
} |
|
} |
|
|
|
// zeroConfig returns a new scrape config that only contains configuration items |
|
// that alter metrics. |
|
func zeroConfig(c *config.ScrapeConfig) *config.ScrapeConfig { |
|
z := *c |
|
// We zero out the fields that for sure don't affect scrape. |
|
z.ScrapeInterval = 0 |
|
z.ScrapeTimeout = 0 |
|
z.SampleLimit = 0 |
|
z.HTTPClientConfig = config_util.HTTPClientConfig{} |
|
return &z |
|
} |
|
|
|
// reusableCache compares two scrape config and tells whether the cache is still |
|
// valid. |
|
func reusableCache(r, l *config.ScrapeConfig) bool { |
|
if r == nil || l == nil { |
|
return false |
|
} |
|
return reflect.DeepEqual(zeroConfig(r), zeroConfig(l)) |
|
} |
|
|
|
// CtxKey is a dedicated type for keys of context-embedded values propagated |
|
// with the scrape context. |
|
type ctxKey int |
|
|
|
// Valid CtxKey values. |
|
const ( |
|
ctxKeyMetadata ctxKey = iota + 1 |
|
ctxKeyTarget |
|
) |
|
|
|
func ContextWithMetricMetadataStore(ctx context.Context, s MetricMetadataStore) context.Context { |
|
return context.WithValue(ctx, ctxKeyMetadata, s) |
|
} |
|
|
|
func MetricMetadataStoreFromContext(ctx context.Context) (MetricMetadataStore, bool) { |
|
s, ok := ctx.Value(ctxKeyMetadata).(MetricMetadataStore) |
|
return s, ok |
|
} |
|
|
|
func ContextWithTarget(ctx context.Context, t *Target) context.Context { |
|
return context.WithValue(ctx, ctxKeyTarget, t) |
|
} |
|
|
|
func TargetFromContext(ctx context.Context) (*Target, bool) { |
|
t, ok := ctx.Value(ctxKeyTarget).(*Target) |
|
return t, ok |
|
}
|
|
|