Scrape Pipeline

Each service discovery mechanism runs as a pluggable Discoverer. The manager aggregates their output and pushes batched updates downstream over a channel.

type Manager struct {
    logger         *slog.Logger
    name           string
    httpClientOpts []config.HTTPClientOption

    // Guards metrics and targetSets.
    mtx sync.RWMutex

    // Sync channel to the consumer (scrape/notify manager).
    syncCh chan map[string][]*targetgroup.Group

    // Providers holds all active SD providers.
    providers []*Provider

    // Target groups keyed by providerRef → jobName.
    targetSets map[string]map[string]*targetgroup.Group
    ...
}

Each SD provider type has its own Discoverer implementation. When target groups change, the provider sends updates into the manager's internal channel, which the manager batches and forwards to syncCh after a configurable updatert delay (default 5 s).

Supported SD Mechanisms

type Manager struct {
    opts         *Options
    logger       *slog.Logger
    append       storage.Appendable
    graceShut    chan struct{}

    offsetSeed    uint64              // used to jitter scrape offsets
    mtxScrape     sync.Mutex         // guards the maps below
    scrapeConfigs map[string]*config.ScrapeConfig
    scrapePools   map[string]*scrapePool  // one pool per job_name
    ...
}

When the Discovery Manager pushes a new set of target groups, Manager.reload() diffs the current and incoming pools:

1. New jobs → create scrapePool, call pool.Sync(tgroups)
2. Removed jobs → call pool.stop() to drain loops
3. Changed targets within a job → pool.Sync() starts/stops individual scrapeLoop goroutines

Target Relabeling

Before a target is scraped, the scrape pool applies relabel.Process() using the job's relabel_configs. Labels prefixed with __ are internal and stripped before storage. Key labels:

func newScrapeLoop(opts scrapeLoopOptions) *scrapeLoop {
    sl := &scrapeLoop{
        target:             opts.target,
        scraper:            opts.scraper,       // HTTP client
        buffers:            pool.New(1e3, 1e6, 3, func(n int) interface{} {
                                return make([]byte, 0, n)
                            }),
        sampleMutator:      opts.sampleMutator,  // relabeling
        reportSampleMutator: opts.reportSampleMutator,
        appender:           opts.appender,        // storage.Appendable
        ...
    }
    return sl
}

func (sl *scrapeLoop) run(errc chan<- error) {
    // Align scrape to a fixed offset within the interval to spread load.
    select {
    case <-time.After(sl.scrapeOffset()):
    case <-sl.ctx.Done():
        return
    }

    ticker := time.NewTicker(sl.interval)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            // 1. HTTP GET target URL
            // 2. Parse response into samples
            // 3. Append to storage
            // 4. Report scrape metadata (__scrape_duration_seconds__, etc.)
            sl.scrapeAndReport(...)
        case <-sl.ctx.Done():
            return
        }
    }
}

Parse & Append Sequence

1. HTTP GET target.URL() with Accept: text/plain; version=0.0.4, application/openmetrics-text
2. Response body piped through textparse.New() or openmetricsparse.New()
3. Each metric family iterated; relabeling applied per sample
4. app.Append(ref, lset, ts, val) called for each float sample
5. app.AppendHistogram() for native histograms
6. app.AppendExemplar() for OpenMetrics exemplars
7. Sample/label limits checked; excess → errSampleLimit
8. app.Commit() — atomically writes all samples of this scrape
9. Stale markers written for series that disappeared since last scrape

Config struct: config/config.go — ScrapeConfig

When a series disappears (target down or metric removed), Prometheus injects a special stale NaN value at the next scrape. This tells the query engine not to extrapolate across the gap.

// For each series seen in the previous scrape but absent now:
app.Append(ref, lset, scrapeTimestamp, math.Float64frombits(value.StaleNaN))