Merge pull request #701 from prometheus/fabxc/decouple
Avoid inter-component blocking if ingestion/scraping blocks
This commit is contained in:
Fabian Reinartz
commit
2843ff6a0d
|
|
@ -19,7 +19,6 @@ import (
|
|||
"math/rand"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
|
@ -53,8 +52,6 @@ const (
|
|||
var (
|
||||
errIngestChannelFull = errors.New("ingestion channel full")
|
||||
|
||||
localhostRepresentations = []string{"127.0.0.1", "localhost"}
|
||||
|
||||
targetIntervalLength = prometheus.NewSummaryVec(
|
||||
prometheus.SummaryOpts{
|
||||
Namespace: namespace,
|
||||
|
|
@ -70,79 +67,84 @@ func init() {
|
|||
prometheus.MustRegister(targetIntervalLength)
|
||||
}
|
||||
|
||||
// TargetState describes the state of a Target.
|
||||
type TargetState int
|
||||
// TargetHealth describes the health state of a target.
|
||||
type TargetHealth int
|
||||
|
||||
func (t TargetState) String() string {
|
||||
func (t TargetHealth) String() string {
|
||||
switch t {
|
||||
case Unknown:
|
||||
case HealthUnknown:
|
||||
return "UNKNOWN"
|
||||
case Healthy:
|
||||
case HealthGood:
|
||||
return "HEALTHY"
|
||||
case Unhealthy:
|
||||
case HealthBad:
|
||||
return "UNHEALTHY"
|
||||
}
|
||||
|
||||
panic("unknown state")
|
||||
}
|
||||
|
||||
const (
|
||||
// Unknown is the state of a Target before it is first scraped.
|
||||
Unknown TargetState = iota
|
||||
HealthUnknown TargetHealth = iota
|
||||
// Healthy is the state of a Target that has been successfully scraped.
|
||||
Healthy
|
||||
HealthGood
|
||||
// Unhealthy is the state of a Target that was scraped unsuccessfully.
|
||||
Unhealthy
|
||||
HealthBad
|
||||
)
|
||||
|
||||
// A Target represents an endpoint that should be interrogated for metrics.
|
||||
//
|
||||
// The protocol described by this type will likely change in future iterations,
|
||||
// as it offers no good support for aggregated targets and fan out. Thusly,
|
||||
// it is likely that the current Target and target uses will be
|
||||
// wrapped with some resolver type.
|
||||
//
|
||||
// For the future, the Target protocol will abstract away the exact means that
|
||||
// metrics are retrieved and deserialized from the given instance to which it
|
||||
// refers.
|
||||
//
|
||||
// Target implements extraction.Ingester.
|
||||
type Target interface {
|
||||
extraction.Ingester
|
||||
// TargetStatus contains information about the current status of a scrape target.
|
||||
type TargetStatus struct {
|
||||
lastError error
|
||||
lastScrape time.Time
|
||||
health TargetHealth
|
||||
|
||||
// Return the last encountered scrape error, if any.
|
||||
LastError() error
|
||||
// Return the health of the target.
|
||||
State() TargetState
|
||||
// Return the last time a scrape was attempted.
|
||||
LastScrape() time.Time
|
||||
// The URL to which the Target corresponds. Out of all of the available
|
||||
// points in this interface, this one is the best candidate to change given
|
||||
// the ways to express the endpoint.
|
||||
URL() string
|
||||
// Used to populate the `instance` label in metrics.
|
||||
InstanceIdentifier() string
|
||||
// The URL as seen from other hosts. References to localhost are resolved
|
||||
// to the address of the prometheus server.
|
||||
GlobalURL() string
|
||||
// Return the labels describing the targets. These are the base labels
|
||||
// as well as internal labels.
|
||||
Labels() clientmodel.LabelSet
|
||||
// Return the target's base labels.
|
||||
BaseLabels() clientmodel.LabelSet
|
||||
// Return the target's base labels without job and instance label. That's
|
||||
// useful for display purposes.
|
||||
BaseLabelsWithoutJobAndInstance() clientmodel.LabelSet
|
||||
// Start scraping the target in regular intervals.
|
||||
RunScraper(storage.SampleAppender)
|
||||
// Stop scraping, synchronous.
|
||||
StopScraper()
|
||||
// Update the target's state.
|
||||
Update(*config.ScrapeConfig, clientmodel.LabelSet)
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// target is a Target that refers to a singular HTTP or HTTPS endpoint.
|
||||
type target struct {
|
||||
// LastError returns the error encountered during the last scrape.
|
||||
func (ts *TargetStatus) LastError() error {
|
||||
ts.mu.RLock()
|
||||
defer ts.mu.RUnlock()
|
||||
return ts.lastError
|
||||
}
|
||||
|
||||
// LastScrape returns the time of the last scrape.
|
||||
func (ts *TargetStatus) LastScrape() time.Time {
|
||||
ts.mu.RLock()
|
||||
defer ts.mu.RUnlock()
|
||||
return ts.lastScrape
|
||||
}
|
||||
|
||||
// Health returns the last known health state of the target.
|
||||
func (ts *TargetStatus) Health() TargetHealth {
|
||||
ts.mu.RLock()
|
||||
defer ts.mu.RUnlock()
|
||||
return ts.health
|
||||
}
|
||||
|
||||
func (ts *TargetStatus) setLastScrape(t time.Time) {
|
||||
ts.mu.Lock()
|
||||
defer ts.mu.Unlock()
|
||||
ts.lastScrape = t
|
||||
}
|
||||
|
||||
func (ts *TargetStatus) setLastError(err error) {
|
||||
ts.mu.Lock()
|
||||
defer ts.mu.Unlock()
|
||||
if err == nil {
|
||||
ts.health = HealthGood
|
||||
} else {
|
||||
ts.health = HealthBad
|
||||
}
|
||||
ts.lastError = err
|
||||
}
|
||||
|
||||
// Target refers to a singular HTTP or HTTPS endpoint.
|
||||
type Target struct {
|
||||
// The status object for the target. It is only set once on initialization.
|
||||
status *TargetStatus
|
||||
// The HTTP client used to scrape the target's endpoint.
|
||||
httpClient *http.Client
|
||||
|
||||
// Closing scraperStopping signals that scraping should stop.
|
||||
scraperStopping chan struct{}
|
||||
// Closing scraperStopped signals that scraping has been stopped.
|
||||
|
|
@ -150,21 +152,12 @@ type target struct {
|
|||
// Channel to buffer ingested samples.
|
||||
ingestedSamples chan clientmodel.Samples
|
||||
|
||||
// The HTTP client used to scrape the target's endpoint.
|
||||
httpClient *http.Client
|
||||
|
||||
// Mutex protects the members below.
|
||||
sync.RWMutex
|
||||
|
||||
// url is the URL to be scraped. Its host is immutable.
|
||||
url *url.URL
|
||||
// Any base labels that are added to this target and its metrics.
|
||||
baseLabels clientmodel.LabelSet
|
||||
// The current health state of the target.
|
||||
state TargetState
|
||||
// The last encountered scrape error, if any.
|
||||
lastError error
|
||||
// The last time a scrape was attempted.
|
||||
lastScrape time.Time
|
||||
// What is the deadline for the HTTP or HTTPS against this endpoint.
|
||||
deadline time.Duration
|
||||
// The time between two scrapes.
|
||||
|
|
@ -172,11 +165,12 @@ type target struct {
|
|||
}
|
||||
|
||||
// NewTarget creates a reasonably configured target for querying.
|
||||
func NewTarget(cfg *config.ScrapeConfig, baseLabels clientmodel.LabelSet) Target {
|
||||
t := &target{
|
||||
func NewTarget(cfg *config.ScrapeConfig, baseLabels clientmodel.LabelSet) *Target {
|
||||
t := &Target{
|
||||
url: &url.URL{
|
||||
Host: string(baseLabels[clientmodel.AddressLabel]),
|
||||
},
|
||||
status: &TargetStatus{},
|
||||
scraperStopping: make(chan struct{}),
|
||||
scraperStopped: make(chan struct{}),
|
||||
}
|
||||
|
|
@ -184,9 +178,14 @@ func NewTarget(cfg *config.ScrapeConfig, baseLabels clientmodel.LabelSet) Target
|
|||
return t
|
||||
}
|
||||
|
||||
// Status returns the status of the target.
|
||||
func (t *Target) Status() *TargetStatus {
|
||||
return t.status
|
||||
}
|
||||
|
||||
// Update overwrites settings in the target that are derived from the job config
|
||||
// it belongs to.
|
||||
func (t *target) Update(cfg *config.ScrapeConfig, baseLabels clientmodel.LabelSet) {
|
||||
func (t *Target) Update(cfg *config.ScrapeConfig, baseLabels clientmodel.LabelSet) {
|
||||
t.Lock()
|
||||
defer t.Unlock()
|
||||
|
||||
|
|
@ -212,12 +211,15 @@ func (t *target) Update(cfg *config.ScrapeConfig, baseLabels clientmodel.LabelSe
|
|||
}
|
||||
}
|
||||
|
||||
func (t *target) String() string {
|
||||
func (t *Target) String() string {
|
||||
return t.url.Host
|
||||
}
|
||||
|
||||
// Ingest implements Target and extraction.Ingester.
|
||||
func (t *target) Ingest(s clientmodel.Samples) error {
|
||||
// Ingest implements an extraction.Ingester.
|
||||
func (t *Target) Ingest(s clientmodel.Samples) error {
|
||||
t.RLock()
|
||||
deadline := t.deadline
|
||||
t.RUnlock()
|
||||
// Since the regular case is that ingestedSamples is ready to receive,
|
||||
// first try without setting a timeout so that we don't need to allocate
|
||||
// a timer most of the time.
|
||||
|
|
@ -228,14 +230,17 @@ func (t *target) Ingest(s clientmodel.Samples) error {
|
|||
select {
|
||||
case t.ingestedSamples <- s:
|
||||
return nil
|
||||
case <-time.After(t.deadline / 10):
|
||||
case <-time.After(deadline / 10):
|
||||
return errIngestChannelFull
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure that Target implements extraction.Ingester at compile time.
|
||||
var _ extraction.Ingester = (*Target)(nil)
|
||||
|
||||
// RunScraper implements Target.
|
||||
func (t *target) RunScraper(sampleAppender storage.SampleAppender) {
|
||||
func (t *Target) RunScraper(sampleAppender storage.SampleAppender) {
|
||||
defer close(t.scraperStopped)
|
||||
|
||||
t.RLock()
|
||||
|
|
@ -256,9 +261,7 @@ func (t *target) RunScraper(sampleAppender storage.SampleAppender) {
|
|||
ticker := time.NewTicker(lastScrapeInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
t.Lock() // Writing t.lastScrape requires the lock.
|
||||
t.lastScrape = time.Now()
|
||||
t.Unlock()
|
||||
t.status.setLastScrape(time.Now())
|
||||
t.scrape(sampleAppender)
|
||||
|
||||
// Explanation of the contraption below:
|
||||
|
|
@ -277,12 +280,12 @@ func (t *target) RunScraper(sampleAppender storage.SampleAppender) {
|
|||
case <-t.scraperStopping:
|
||||
return
|
||||
case <-ticker.C:
|
||||
t.Lock()
|
||||
took := time.Since(t.lastScrape)
|
||||
t.lastScrape = time.Now()
|
||||
took := time.Since(t.status.LastScrape())
|
||||
t.status.setLastScrape(time.Now())
|
||||
|
||||
intervalStr := lastScrapeInterval.String()
|
||||
|
||||
t.RLock()
|
||||
// On changed scrape interval the new interval becomes effective
|
||||
// after the next scrape.
|
||||
if lastScrapeInterval != t.scrapeInterval {
|
||||
|
|
@ -290,7 +293,7 @@ func (t *target) RunScraper(sampleAppender storage.SampleAppender) {
|
|||
ticker = time.NewTicker(t.scrapeInterval)
|
||||
lastScrapeInterval = t.scrapeInterval
|
||||
}
|
||||
t.Unlock()
|
||||
t.RUnlock()
|
||||
|
||||
targetIntervalLength.WithLabelValues(intervalStr).Observe(
|
||||
float64(took) / float64(time.Second), // Sub-second precision.
|
||||
|
|
@ -302,7 +305,7 @@ func (t *target) RunScraper(sampleAppender storage.SampleAppender) {
|
|||
}
|
||||
|
||||
// StopScraper implements Target.
|
||||
func (t *target) StopScraper() {
|
||||
func (t *Target) StopScraper() {
|
||||
glog.V(1).Infof("Stopping scraper for target %v...", t)
|
||||
|
||||
close(t.scraperStopping)
|
||||
|
|
@ -313,25 +316,16 @@ func (t *target) StopScraper() {
|
|||
|
||||
const acceptHeader = `application/vnd.google.protobuf;proto=io.prometheus.client.MetricFamily;encoding=delimited;q=0.7,text/plain;version=0.0.4;q=0.3,application/json;schema="prometheus/telemetry";version=0.0.2;q=0.2,*/*;q=0.1`
|
||||
|
||||
func (t *target) scrape(sampleAppender storage.SampleAppender) (err error) {
|
||||
t.RLock()
|
||||
timestamp := clientmodel.Now()
|
||||
func (t *Target) scrape(sampleAppender storage.SampleAppender) (err error) {
|
||||
start := time.Now()
|
||||
baseLabels := t.BaseLabels()
|
||||
|
||||
defer func(start time.Time) {
|
||||
t.recordScrapeHealth(sampleAppender, timestamp, err == nil, time.Since(start))
|
||||
t.RUnlock()
|
||||
defer func() {
|
||||
t.status.setLastError(err)
|
||||
recordScrapeHealth(sampleAppender, clientmodel.TimestampFromTime(start), baseLabels, t.status.Health(), time.Since(start))
|
||||
}()
|
||||
|
||||
t.Lock() // Writing t.state and t.lastError requires the lock.
|
||||
if err == nil {
|
||||
t.state = Healthy
|
||||
} else {
|
||||
t.state = Unhealthy
|
||||
}
|
||||
t.lastError = err
|
||||
t.Unlock()
|
||||
}(time.Now())
|
||||
|
||||
req, err := http.NewRequest("GET", t.url.String(), nil)
|
||||
req, err := http.NewRequest("GET", t.URL(), nil)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
|
@ -354,7 +348,7 @@ func (t *target) scrape(sampleAppender storage.SampleAppender) (err error) {
|
|||
t.ingestedSamples = make(chan clientmodel.Samples, ingestedSamplesCap)
|
||||
|
||||
processOptions := &extraction.ProcessOptions{
|
||||
Timestamp: timestamp,
|
||||
Timestamp: clientmodel.TimestampFromTime(start),
|
||||
}
|
||||
go func() {
|
||||
err = processor.ProcessSingle(resp.Body, t, processOptions)
|
||||
|
|
@ -363,109 +357,69 @@ func (t *target) scrape(sampleAppender storage.SampleAppender) (err error) {
|
|||
|
||||
for samples := range t.ingestedSamples {
|
||||
for _, s := range samples {
|
||||
s.Metric.MergeFromLabelSet(t.baseLabels, clientmodel.ExporterLabelPrefix)
|
||||
s.Metric.MergeFromLabelSet(baseLabels, clientmodel.ExporterLabelPrefix)
|
||||
sampleAppender.Append(s)
|
||||
}
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// LastError implements Target.
|
||||
func (t *target) LastError() error {
|
||||
t.RLock()
|
||||
defer t.RUnlock()
|
||||
return t.lastError
|
||||
}
|
||||
|
||||
// State implements Target.
|
||||
func (t *target) State() TargetState {
|
||||
t.RLock()
|
||||
defer t.RUnlock()
|
||||
return t.state
|
||||
}
|
||||
|
||||
// LastScrape implements Target.
|
||||
func (t *target) LastScrape() time.Time {
|
||||
t.RLock()
|
||||
defer t.RUnlock()
|
||||
return t.lastScrape
|
||||
}
|
||||
|
||||
// URL implements Target.
|
||||
func (t *target) URL() string {
|
||||
func (t *Target) URL() string {
|
||||
t.RLock()
|
||||
defer t.RUnlock()
|
||||
return t.url.String()
|
||||
}
|
||||
|
||||
// InstanceIdentifier implements Target.
|
||||
func (t *target) InstanceIdentifier() string {
|
||||
// InstanceIdentifier returns the identifier for the target.
|
||||
func (t *Target) InstanceIdentifier() string {
|
||||
return t.url.Host
|
||||
}
|
||||
|
||||
// GlobalURL implements Target.
|
||||
func (t *target) GlobalURL() string {
|
||||
url := t.URL()
|
||||
|
||||
hostname, err := os.Hostname()
|
||||
if err != nil {
|
||||
glog.Warningf("Couldn't get hostname: %s, returning target.URL()", err)
|
||||
return url
|
||||
}
|
||||
for _, localhostRepresentation := range localhostRepresentations {
|
||||
url = strings.Replace(url, "//"+localhostRepresentation, "//"+hostname, 1)
|
||||
}
|
||||
return url
|
||||
}
|
||||
|
||||
// Labels implements Target.
|
||||
func (t *target) Labels() clientmodel.LabelSet {
|
||||
// fullLabels returns the base labels plus internal labels defining the target.
|
||||
func (t *Target) fullLabels() clientmodel.LabelSet {
|
||||
t.RLock()
|
||||
defer t.RUnlock()
|
||||
ls := clientmodel.LabelSet{}
|
||||
lset := make(clientmodel.LabelSet, len(t.baseLabels)+2)
|
||||
for ln, lv := range t.baseLabels {
|
||||
ls[ln] = lv
|
||||
lset[ln] = lv
|
||||
}
|
||||
ls[clientmodel.MetricsPathLabel] = clientmodel.LabelValue(t.url.Path)
|
||||
ls[clientmodel.AddressLabel] = clientmodel.LabelValue(t.url.Host)
|
||||
return ls
|
||||
lset[clientmodel.MetricsPathLabel] = clientmodel.LabelValue(t.url.Path)
|
||||
lset[clientmodel.AddressLabel] = clientmodel.LabelValue(t.url.Host)
|
||||
return lset
|
||||
}
|
||||
|
||||
// BaseLabels implements Target.
|
||||
func (t *target) BaseLabels() clientmodel.LabelSet {
|
||||
// BaseLabels returns a copy of the target's base labels.
|
||||
func (t *Target) BaseLabels() clientmodel.LabelSet {
|
||||
t.RLock()
|
||||
defer t.RUnlock()
|
||||
return t.baseLabels
|
||||
}
|
||||
|
||||
// BaseLabelsWithoutJobAndInstance implements Target.
|
||||
//
|
||||
// TODO(fabxc): This method does not have to be part of the interface. Implement this
|
||||
// as a template filter func for the single use case.
|
||||
func (t *target) BaseLabelsWithoutJobAndInstance() clientmodel.LabelSet {
|
||||
t.RLock()
|
||||
defer t.RUnlock()
|
||||
ls := clientmodel.LabelSet{}
|
||||
lset := make(clientmodel.LabelSet, len(t.baseLabels))
|
||||
for ln, lv := range t.baseLabels {
|
||||
if ln != clientmodel.JobLabel && ln != clientmodel.InstanceLabel {
|
||||
ls[ln] = lv
|
||||
}
|
||||
lset[ln] = lv
|
||||
}
|
||||
return ls
|
||||
return lset
|
||||
}
|
||||
|
||||
func (t *target) recordScrapeHealth(sampleAppender storage.SampleAppender, timestamp clientmodel.Timestamp, healthy bool, scrapeDuration time.Duration) {
|
||||
healthMetric := clientmodel.Metric{}
|
||||
durationMetric := clientmodel.Metric{}
|
||||
for label, value := range t.baseLabels {
|
||||
healthMetric[label] = value
|
||||
durationMetric[label] = value
|
||||
}
|
||||
func recordScrapeHealth(
|
||||
sampleAppender storage.SampleAppender,
|
||||
timestamp clientmodel.Timestamp,
|
||||
baseLabels clientmodel.LabelSet,
|
||||
health TargetHealth,
|
||||
scrapeDuration time.Duration,
|
||||
) {
|
||||
healthMetric := make(clientmodel.Metric, len(baseLabels)+1)
|
||||
durationMetric := make(clientmodel.Metric, len(baseLabels)+1)
|
||||
|
||||
healthMetric[clientmodel.MetricNameLabel] = clientmodel.LabelValue(scrapeHealthMetricName)
|
||||
durationMetric[clientmodel.MetricNameLabel] = clientmodel.LabelValue(scrapeDurationMetricName)
|
||||
|
||||
for label, value := range baseLabels {
|
||||
healthMetric[label] = value
|
||||
durationMetric[label] = value
|
||||
}
|
||||
|
||||
healthValue := clientmodel.SampleValue(0)
|
||||
if healthy {
|
||||
if health == HealthGood {
|
||||
healthValue = clientmodel.SampleValue(1)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -29,10 +29,6 @@ import (
|
|||
"github.com/prometheus/prometheus/utility"
|
||||
)
|
||||
|
||||
func TestTargetInterface(t *testing.T) {
|
||||
var _ Target = &target{}
|
||||
}
|
||||
|
||||
func TestBaseLabels(t *testing.T) {
|
||||
target := newTestTarget("example.com:80", 0, clientmodel.LabelSet{"job": "some_job", "foo": "bar"})
|
||||
want := clientmodel.LabelSet{
|
||||
|
|
@ -44,21 +40,14 @@ func TestBaseLabels(t *testing.T) {
|
|||
if !reflect.DeepEqual(want, got) {
|
||||
t.Errorf("want base labels %v, got %v", want, got)
|
||||
}
|
||||
delete(want, clientmodel.JobLabel)
|
||||
delete(want, clientmodel.InstanceLabel)
|
||||
|
||||
got = target.BaseLabelsWithoutJobAndInstance()
|
||||
if !reflect.DeepEqual(want, got) {
|
||||
t.Errorf("want base labels %v, got %v", want, got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTargetScrapeUpdatesState(t *testing.T) {
|
||||
testTarget := newTestTarget("bad schema", 0, nil)
|
||||
|
||||
testTarget.scrape(nopAppender{})
|
||||
if testTarget.state != Unhealthy {
|
||||
t.Errorf("Expected target state %v, actual: %v", Unhealthy, testTarget.state)
|
||||
if testTarget.status.Health() != HealthBad {
|
||||
t.Errorf("Expected target state %v, actual: %v", HealthBad, testTarget.status.Health())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -80,11 +69,11 @@ func TestTargetScrapeWithFullChannel(t *testing.T) {
|
|||
testTarget := newTestTarget(server.URL, 10*time.Millisecond, clientmodel.LabelSet{"dings": "bums"})
|
||||
|
||||
testTarget.scrape(slowAppender{})
|
||||
if testTarget.state != Unhealthy {
|
||||
t.Errorf("Expected target state %v, actual: %v", Unhealthy, testTarget.state)
|
||||
if testTarget.status.Health() != HealthBad {
|
||||
t.Errorf("Expected target state %v, actual: %v", HealthBad, testTarget.status.Health())
|
||||
}
|
||||
if testTarget.lastError != errIngestChannelFull {
|
||||
t.Errorf("Expected target error %q, actual: %q", errIngestChannelFull, testTarget.lastError)
|
||||
if testTarget.status.LastError() != errIngestChannelFull {
|
||||
t.Errorf("Expected target error %q, actual: %q", errIngestChannelFull, testTarget.status.LastError())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -93,7 +82,8 @@ func TestTargetRecordScrapeHealth(t *testing.T) {
|
|||
|
||||
now := clientmodel.Now()
|
||||
appender := &collectResultAppender{}
|
||||
testTarget.recordScrapeHealth(appender, now, true, 2*time.Second)
|
||||
testTarget.status.setLastError(nil)
|
||||
recordScrapeHealth(appender, now, testTarget.BaseLabels(), testTarget.status.Health(), 2*time.Second)
|
||||
|
||||
result := appender.result
|
||||
|
||||
|
|
@ -145,13 +135,13 @@ func TestTargetScrapeTimeout(t *testing.T) {
|
|||
)
|
||||
defer server.Close()
|
||||
|
||||
var testTarget Target = newTestTarget(server.URL, 10*time.Millisecond, clientmodel.LabelSet{})
|
||||
testTarget := newTestTarget(server.URL, 10*time.Millisecond, clientmodel.LabelSet{})
|
||||
|
||||
appender := nopAppender{}
|
||||
|
||||
// scrape once without timeout
|
||||
signal <- true
|
||||
if err := testTarget.(*target).scrape(appender); err != nil {
|
||||
if err := testTarget.scrape(appender); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -160,12 +150,12 @@ func TestTargetScrapeTimeout(t *testing.T) {
|
|||
|
||||
// now scrape again
|
||||
signal <- true
|
||||
if err := testTarget.(*target).scrape(appender); err != nil {
|
||||
if err := testTarget.scrape(appender); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// now timeout
|
||||
if err := testTarget.(*target).scrape(appender); err == nil {
|
||||
if err := testTarget.scrape(appender); err == nil {
|
||||
t.Fatal("expected scrape to timeout")
|
||||
} else {
|
||||
signal <- true // let handler continue
|
||||
|
|
@ -173,7 +163,7 @@ func TestTargetScrapeTimeout(t *testing.T) {
|
|||
|
||||
// now scrape again without timeout
|
||||
signal <- true
|
||||
if err := testTarget.(*target).scrape(appender); err != nil {
|
||||
if err := testTarget.scrape(appender); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
|
@ -205,17 +195,17 @@ func TestTargetRunScraperScrapes(t *testing.T) {
|
|||
|
||||
// Enough time for a scrape to happen.
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
if testTarget.lastScrape.IsZero() {
|
||||
if testTarget.status.LastScrape().IsZero() {
|
||||
t.Errorf("Scrape hasn't occured.")
|
||||
}
|
||||
|
||||
testTarget.StopScraper()
|
||||
// Wait for it to take effect.
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
last := testTarget.lastScrape
|
||||
last := testTarget.status.LastScrape()
|
||||
// Enough time for a scrape to happen.
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
if testTarget.lastScrape != last {
|
||||
if testTarget.status.LastScrape() != last {
|
||||
t.Errorf("Scrape occured after it was stopped.")
|
||||
}
|
||||
}
|
||||
|
|
@ -231,25 +221,26 @@ func BenchmarkScrape(b *testing.B) {
|
|||
)
|
||||
defer server.Close()
|
||||
|
||||
var testTarget Target = newTestTarget(server.URL, 100*time.Millisecond, clientmodel.LabelSet{"dings": "bums"})
|
||||
testTarget := newTestTarget(server.URL, 100*time.Millisecond, clientmodel.LabelSet{"dings": "bums"})
|
||||
appender := nopAppender{}
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
if err := testTarget.(*target).scrape(appender); err != nil {
|
||||
if err := testTarget.scrape(appender); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func newTestTarget(targetURL string, deadline time.Duration, baseLabels clientmodel.LabelSet) *target {
|
||||
t := &target{
|
||||
func newTestTarget(targetURL string, deadline time.Duration, baseLabels clientmodel.LabelSet) *Target {
|
||||
t := &Target{
|
||||
url: &url.URL{
|
||||
Scheme: "http",
|
||||
Host: strings.TrimLeft(targetURL, "http://"),
|
||||
Path: "/metrics",
|
||||
},
|
||||
deadline: deadline,
|
||||
status: &TargetStatus{},
|
||||
scrapeInterval: 1 * time.Millisecond,
|
||||
httpClient: utility.NewDeadlineClient(deadline),
|
||||
scraperStopping: make(chan struct{}),
|
||||
|
|
|
|||
|
|
@ -56,7 +56,7 @@ type TargetManager struct {
|
|||
running bool
|
||||
|
||||
// Targets by their source ID.
|
||||
targets map[string][]Target
|
||||
targets map[string][]*Target
|
||||
// Providers by the scrape configs they are derived from.
|
||||
providers map[*config.ScrapeConfig][]TargetProvider
|
||||
}
|
||||
|
|
@ -65,7 +65,7 @@ type TargetManager struct {
|
|||
func NewTargetManager(sampleAppender storage.SampleAppender) *TargetManager {
|
||||
tm := &TargetManager{
|
||||
sampleAppender: sampleAppender,
|
||||
targets: make(map[string][]Target),
|
||||
targets: make(map[string][]*Target),
|
||||
}
|
||||
return tm
|
||||
}
|
||||
|
|
@ -165,7 +165,7 @@ func (tm *TargetManager) removeTargets(f func(string) bool) {
|
|||
}
|
||||
wg.Add(len(targets))
|
||||
for _, target := range targets {
|
||||
go func(t Target) {
|
||||
go func(t *Target) {
|
||||
t.StopScraper()
|
||||
wg.Done()
|
||||
}(target)
|
||||
|
|
@ -197,7 +197,7 @@ func (tm *TargetManager) updateTargetGroup(tgroup *config.TargetGroup, cfg *conf
|
|||
// Replace the old targets with the new ones while keeping the state
|
||||
// of intersecting targets.
|
||||
for i, tnew := range newTargets {
|
||||
var match Target
|
||||
var match *Target
|
||||
for j, told := range oldTargets {
|
||||
if told == nil {
|
||||
continue
|
||||
|
|
@ -214,8 +214,8 @@ func (tm *TargetManager) updateTargetGroup(tgroup *config.TargetGroup, cfg *conf
|
|||
// Updating is blocked during a scrape. We don't want those wait times
|
||||
// to build up.
|
||||
wg.Add(1)
|
||||
go func(t Target) {
|
||||
match.Update(cfg, t.Labels())
|
||||
go func(t *Target) {
|
||||
match.Update(cfg, t.fullLabels())
|
||||
wg.Done()
|
||||
}(tnew)
|
||||
newTargets[i] = match
|
||||
|
|
@ -227,7 +227,7 @@ func (tm *TargetManager) updateTargetGroup(tgroup *config.TargetGroup, cfg *conf
|
|||
for _, told := range oldTargets {
|
||||
if told != nil {
|
||||
wg.Add(1)
|
||||
go func(t Target) {
|
||||
go func(t *Target) {
|
||||
t.StopScraper()
|
||||
wg.Done()
|
||||
}(told)
|
||||
|
|
@ -250,11 +250,11 @@ func (tm *TargetManager) updateTargetGroup(tgroup *config.TargetGroup, cfg *conf
|
|||
}
|
||||
|
||||
// Pools returns the targets currently being scraped bucketed by their job name.
|
||||
func (tm *TargetManager) Pools() map[string][]Target {
|
||||
func (tm *TargetManager) Pools() map[string][]*Target {
|
||||
tm.m.RLock()
|
||||
defer tm.m.RUnlock()
|
||||
|
||||
pools := map[string][]Target{}
|
||||
pools := map[string][]*Target{}
|
||||
|
||||
for _, ts := range tm.targets {
|
||||
for _, t := range ts {
|
||||
|
|
@ -287,11 +287,11 @@ func (tm *TargetManager) ApplyConfig(cfg *config.Config) {
|
|||
}
|
||||
|
||||
// targetsFromGroup builds targets based on the given TargetGroup and config.
|
||||
func (tm *TargetManager) targetsFromGroup(tg *config.TargetGroup, cfg *config.ScrapeConfig) ([]Target, error) {
|
||||
func (tm *TargetManager) targetsFromGroup(tg *config.TargetGroup, cfg *config.ScrapeConfig) ([]*Target, error) {
|
||||
tm.m.RLock()
|
||||
defer tm.m.RUnlock()
|
||||
|
||||
targets := make([]Target, 0, len(tg.Targets))
|
||||
targets := make([]*Target, 0, len(tg.Targets))
|
||||
for i, labels := range tg.Targets {
|
||||
addr := string(labels[clientmodel.AddressLabel])
|
||||
// If no port was provided, infer it based on the used scheme.
|
||||
|
|
|
|||
|
|
@ -45,7 +45,7 @@ func TestTargetManagerChan(t *testing.T) {
|
|||
providers: map[*config.ScrapeConfig][]TargetProvider{
|
||||
testJob1: []TargetProvider{prov1},
|
||||
},
|
||||
targets: make(map[string][]Target),
|
||||
targets: make(map[string][]*Target),
|
||||
}
|
||||
go targetManager.Run()
|
||||
defer targetManager.Stop()
|
||||
|
|
|
|||
|
|
@ -32,18 +32,18 @@ type PrometheusStatusHandler struct {
|
|||
Flags map[string]string
|
||||
|
||||
RuleManager *rules.Manager
|
||||
TargetPools func() map[string][]retrieval.Target
|
||||
TargetPools func() map[string][]*retrieval.Target
|
||||
|
||||
Birth time.Time
|
||||
PathPrefix string
|
||||
}
|
||||
|
||||
// TargetStateToClass returns a map of TargetState to the name of a Bootstrap CSS class.
|
||||
func (h *PrometheusStatusHandler) TargetStateToClass() map[retrieval.TargetState]string {
|
||||
return map[retrieval.TargetState]string{
|
||||
retrieval.Unknown: "warning",
|
||||
retrieval.Healthy: "success",
|
||||
retrieval.Unhealthy: "danger",
|
||||
// TargetHealthToClass returns a map of TargetHealth to the name of a Bootstrap CSS class.
|
||||
func (h *PrometheusStatusHandler) TargetHealthToClass() map[retrieval.TargetHealth]string {
|
||||
return map[retrieval.TargetHealth]string{
|
||||
retrieval.HealthUnknown: "warning",
|
||||
retrieval.HealthBad: "success",
|
||||
retrieval.HealthGood: "danger",
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -32,7 +32,6 @@
|
|||
|
||||
<h2>Targets</h2>
|
||||
<table class="table table-condensed table-bordered table-striped table-hover">
|
||||
{{$stateToClass := .TargetStateToClass}}
|
||||
{{range $job, $pool := call .TargetPools}}
|
||||
<thead>
|
||||
<tr><th colspan="5" class="job_header">{{$job}}</th></tr>
|
||||
|
|
@ -48,22 +47,22 @@
|
|||
{{range $pool}}
|
||||
<tr>
|
||||
<td>
|
||||
<a href="{{.GlobalURL}}">{{.URL}}</a>
|
||||
<a href="{{.URL | globalURL}}">{{.URL}}</a>
|
||||
</td>
|
||||
<td>
|
||||
<span class="alert alert-{{index $stateToClass .State}} target_status_alert">
|
||||
{{.State}}
|
||||
<span class="alert alert-{{index .TargetHealthToClass .Status.State}} target_status_alert">
|
||||
{{.Status.State}}
|
||||
</span>
|
||||
</td>
|
||||
<td>
|
||||
{{.BaseLabelsWithoutJobAndInstance}}
|
||||
{{stripLabels .BaseLabels "job" "instance"}}
|
||||
</td>
|
||||
<td>
|
||||
{{if .LastScrape.IsZero}}Never{{else}}{{since .LastScrape}} ago{{end}}
|
||||
{{if .Status.LastScrape.IsZero}}Never{{else}}{{since .Status.LastScrape}} ago{{end}}
|
||||
</td>
|
||||
<td>
|
||||
{{if .LastError}}
|
||||
<span class="alert alert-danger target_status_alert">{{.LastError}}</span>
|
||||
{{if .Status.LastError}}
|
||||
<span class="alert alert-danger target_status_alert">{{.Status.LastError}}</span>
|
||||
{{end}}
|
||||
</td>
|
||||
</tr>
|
||||
|
|
|
|||
43
web/web.go
43
web/web.go
|
|
@ -29,10 +29,14 @@ import (
|
|||
"github.com/golang/glog"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
clientmodel "github.com/prometheus/client_golang/model"
|
||||
|
||||
"github.com/prometheus/prometheus/web/api"
|
||||
"github.com/prometheus/prometheus/web/blob"
|
||||
)
|
||||
|
||||
var localhostRepresentations = []string{"127.0.0.1", "localhost"}
|
||||
|
||||
// Commandline flags.
|
||||
var (
|
||||
listenAddress = flag.String("web.listen-address", ":9090", "Address to listen on for the web interface, API, and telemetry.")
|
||||
|
|
@ -150,28 +154,53 @@ func getConsoles(pathPrefix string) string {
|
|||
return ""
|
||||
}
|
||||
|
||||
func getTemplate(name string, pathPrefix string) (t *template.Template, err error) {
|
||||
t = template.New("_base")
|
||||
func getTemplate(name string, pathPrefix string) (*template.Template, error) {
|
||||
t := template.New("_base")
|
||||
var err error
|
||||
|
||||
t.Funcs(template.FuncMap{
|
||||
"since": time.Since,
|
||||
"getConsoles": func() string { return getConsoles(pathPrefix) },
|
||||
"pathPrefix": func() string { return pathPrefix },
|
||||
"stripLabels": func(lset clientmodel.LabelSet, labels ...clientmodel.LabelName) clientmodel.LabelSet {
|
||||
for _, ln := range labels {
|
||||
delete(lset, ln)
|
||||
}
|
||||
return lset
|
||||
},
|
||||
"globalURL": func(url string) string {
|
||||
hostname, err := os.Hostname()
|
||||
if err != nil {
|
||||
glog.Warningf("Couldn't get hostname: %s, returning target.URL()", err)
|
||||
return url
|
||||
}
|
||||
for _, localhostRepresentation := range localhostRepresentations {
|
||||
url = strings.Replace(url, "//"+localhostRepresentation, "//"+hostname, 1)
|
||||
}
|
||||
return url
|
||||
},
|
||||
})
|
||||
|
||||
file, err := getTemplateFile("_base")
|
||||
if err != nil {
|
||||
glog.Error("Could not read base template: ", err)
|
||||
glog.Errorln("Could not read base template:", err)
|
||||
return nil, err
|
||||
}
|
||||
t.Parse(file)
|
||||
t, err = t.Parse(file)
|
||||
if err != nil {
|
||||
glog.Errorln("Could not parse base template:", err)
|
||||
}
|
||||
|
||||
file, err = getTemplateFile(name)
|
||||
if err != nil {
|
||||
glog.Error("Could not read base template: ", err)
|
||||
glog.Error("Could not read template %s: %s", name, err)
|
||||
return nil, err
|
||||
}
|
||||
t.Parse(file)
|
||||
return
|
||||
t, err = t.Parse(file)
|
||||
if err != nil {
|
||||
glog.Errorf("Could not parse template %s: %s", name, err)
|
||||
}
|
||||
return t, err
|
||||
}
|
||||
|
||||
func executeTemplate(w http.ResponseWriter, name string, data interface{}, pathPrefix string) {
|
||||
|
|
|
|||
Loading…
Reference in New Issue