mirror of
https://github.com/prometheus/prometheus.git
synced 2024-12-27 14:39:40 -08:00
Merge "Remove backoff on scrape failure."
This commit is contained in:
commit
ffa4a2935e
|
@ -14,20 +14,9 @@
|
||||||
package retrieval
|
package retrieval
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/prometheus/client_golang/extraction"
|
"github.com/prometheus/client_golang/extraction"
|
||||||
)
|
)
|
||||||
|
|
||||||
type literalScheduler time.Time
|
|
||||||
|
|
||||||
func (s literalScheduler) ScheduledFor() time.Time {
|
|
||||||
return time.Time(s)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s literalScheduler) Reschedule(earliest time.Time, future TargetState) {
|
|
||||||
}
|
|
||||||
|
|
||||||
type nopIngester struct{}
|
type nopIngester struct{}
|
||||||
|
|
||||||
func (i nopIngester) Ingest(*extraction.Result) error {
|
func (i nopIngester) Ingest(*extraction.Result) error {
|
||||||
|
|
|
@ -19,9 +19,7 @@ import (
|
||||||
|
|
||||||
func TestInterfaces(t *testing.T) {
|
func TestInterfaces(t *testing.T) {
|
||||||
var (
|
var (
|
||||||
_ Target = &target{}
|
_ Target = &target{}
|
||||||
_ TargetManager = &targetManager{}
|
_ TargetManager = &targetManager{}
|
||||||
_ healthReporter = &target{}
|
|
||||||
_ scheduler = &healthScheduler{}
|
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,118 +0,0 @@
|
||||||
// Copyright 2013 Prometheus Team
|
|
||||||
// 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 retrieval
|
|
||||||
|
|
||||||
import (
|
|
||||||
"github.com/prometheus/prometheus/utility"
|
|
||||||
"math"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
const (
|
|
||||||
// The default increment for exponential backoff when querying a target.
|
|
||||||
DEFAULT_BACKOFF_VALUE = 2
|
|
||||||
// The base units for the exponential backoff.
|
|
||||||
DEFAULT_BACKOFF_VALUE_UNIT = time.Second
|
|
||||||
// The maximum allowed backoff time.
|
|
||||||
MAXIMUM_BACKOFF_VALUE = 2 * time.Minute
|
|
||||||
)
|
|
||||||
|
|
||||||
// scheduler is an interface that various scheduling strategies must fulfill
|
|
||||||
// in order to set the scheduling order for a target.
|
|
||||||
//
|
|
||||||
// Target takes advantage of this type by embedding an instance of scheduler
|
|
||||||
// in each Target instance itself. The emitted scheduler.ScheduledFor() is
|
|
||||||
// the basis for sorting the order of pending queries.
|
|
||||||
//
|
|
||||||
// This type is described as an interface to maximize testability.
|
|
||||||
type scheduler interface {
|
|
||||||
// ScheduledFor emits the earliest time at which the given object is allowed
|
|
||||||
// to be run. This time may or not be a reflection of the earliest parameter
|
|
||||||
// provided in Reschedule; that is up to the underlying strategy
|
|
||||||
// implementations.
|
|
||||||
ScheduledFor() time.Time
|
|
||||||
// Instruct the scheduled item to re-schedule itself given new state data and
|
|
||||||
// the earliest time at which the outside system thinks the operation should
|
|
||||||
// be scheduled for.
|
|
||||||
Reschedule(earliest time.Time, future TargetState)
|
|
||||||
}
|
|
||||||
|
|
||||||
// healthScheduler is an implementation of scheduler that uses health data
|
|
||||||
// provided by the target field as well as unreachability counts to determine
|
|
||||||
// when to next schedule an operation.
|
|
||||||
//
|
|
||||||
// The type is almost capable of being used with default initialization, except
|
|
||||||
// that a target field must be provided for which the system compares current
|
|
||||||
// health against future proposed values.
|
|
||||||
type healthScheduler struct {
|
|
||||||
scheduledFor time.Time
|
|
||||||
target healthReporter
|
|
||||||
time utility.Time
|
|
||||||
unreachableCount int
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s healthScheduler) ScheduledFor() time.Time {
|
|
||||||
return s.scheduledFor
|
|
||||||
}
|
|
||||||
|
|
||||||
// Reschedule, like the protocol described in scheduler, uses the current and
|
|
||||||
// proposed future health state to determine how and when a given subject is to
|
|
||||||
// be scheduled.
|
|
||||||
//
|
|
||||||
// If a subject has been at given moment marked as unhealthy, an exponential
|
|
||||||
// backoff scheme is applied to it. The reason for this backoff is to ensure
|
|
||||||
// that known-healthy targets can consume valuable request queuing resources
|
|
||||||
// first. Depending on the retrieval interval and number of consecutive
|
|
||||||
// unhealthy markings, the query of these unhealthy individuals may come before
|
|
||||||
// the healthy ones for a short time to help ensure expeditious retrieval.
|
|
||||||
// The inflection point that drops these to the back of the queue is beneficial
|
|
||||||
// to save resources in the long-run.
|
|
||||||
//
|
|
||||||
// If a subject is healthy, its next scheduling opportunity is set to
|
|
||||||
// earliest, for this ensures fair querying of all remaining healthy targets and
|
|
||||||
// removes bias in the ordering. In order for the anti-bias value to have any
|
|
||||||
// value, the earliest opportunity should be set to a value that is constant
|
|
||||||
// for a given batch of subjects who are to be scraped on a given interval.
|
|
||||||
func (s *healthScheduler) Reschedule(e time.Time, f TargetState) {
|
|
||||||
currentState := s.target.State()
|
|
||||||
// XXX: Handle metrics surrounding health.
|
|
||||||
switch currentState {
|
|
||||||
case UNKNOWN, UNREACHABLE:
|
|
||||||
switch f {
|
|
||||||
case ALIVE:
|
|
||||||
s.unreachableCount = 0
|
|
||||||
break
|
|
||||||
case UNREACHABLE:
|
|
||||||
s.unreachableCount++
|
|
||||||
break
|
|
||||||
}
|
|
||||||
case ALIVE:
|
|
||||||
switch f {
|
|
||||||
case UNREACHABLE:
|
|
||||||
s.unreachableCount++
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if s.unreachableCount == 0 {
|
|
||||||
s.scheduledFor = e
|
|
||||||
} else {
|
|
||||||
backoff := MAXIMUM_BACKOFF_VALUE
|
|
||||||
exponential := time.Duration(math.Pow(DEFAULT_BACKOFF_VALUE, float64(s.unreachableCount))) * DEFAULT_BACKOFF_VALUE_UNIT
|
|
||||||
if backoff > exponential {
|
|
||||||
backoff = exponential
|
|
||||||
}
|
|
||||||
|
|
||||||
s.scheduledFor = s.time.Now().Add(backoff)
|
|
||||||
}
|
|
||||||
}
|
|
|
@ -1,141 +0,0 @@
|
||||||
// Copyright 2013 Prometheus Team
|
|
||||||
// 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 retrieval
|
|
||||||
|
|
||||||
import (
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/prometheus/prometheus/utility"
|
|
||||||
"github.com/prometheus/prometheus/utility/test"
|
|
||||||
)
|
|
||||||
|
|
||||||
type fakeHealthReporter struct {
|
|
||||||
index int
|
|
||||||
stateQueue []TargetState
|
|
||||||
}
|
|
||||||
|
|
||||||
func (h fakeHealthReporter) State() (state TargetState) {
|
|
||||||
state = h.stateQueue[h.index]
|
|
||||||
|
|
||||||
h.index++
|
|
||||||
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
func testHealthScheduler(t testing.TB) {
|
|
||||||
now := time.Now()
|
|
||||||
var scenarios = []struct {
|
|
||||||
futureHealthState []TargetState
|
|
||||||
preloadedTimes []time.Time
|
|
||||||
expectedSchedule []time.Time
|
|
||||||
}{
|
|
||||||
// The behavior discussed in healthScheduler.Reschedule should be read
|
|
||||||
// fully to understand the whys and wherefores.
|
|
||||||
{
|
|
||||||
futureHealthState: []TargetState{UNKNOWN, ALIVE, ALIVE},
|
|
||||||
preloadedTimes: []time.Time{now, now.Add(time.Minute), now.Add(time.Minute * 2)},
|
|
||||||
expectedSchedule: []time.Time{now, now.Add(time.Minute), now.Add(time.Minute * 2)},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
futureHealthState: []TargetState{UNKNOWN, UNREACHABLE, UNREACHABLE},
|
|
||||||
preloadedTimes: []time.Time{now, now.Add(time.Minute), now.Add(time.Minute * 2)},
|
|
||||||
expectedSchedule: []time.Time{now, now.Add(time.Second * 2), now.Add(time.Minute).Add(time.Second * 4)},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
futureHealthState: []TargetState{UNKNOWN, UNREACHABLE, ALIVE},
|
|
||||||
preloadedTimes: []time.Time{now, now.Add(time.Minute), now.Add(time.Minute * 2)},
|
|
||||||
expectedSchedule: []time.Time{now, now.Add(time.Second * 2), now.Add(time.Minute * 2)},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
futureHealthState: []TargetState{
|
|
||||||
UNKNOWN,
|
|
||||||
UNREACHABLE,
|
|
||||||
UNREACHABLE,
|
|
||||||
UNREACHABLE,
|
|
||||||
UNREACHABLE,
|
|
||||||
UNREACHABLE,
|
|
||||||
UNREACHABLE,
|
|
||||||
UNREACHABLE,
|
|
||||||
UNREACHABLE,
|
|
||||||
UNREACHABLE,
|
|
||||||
},
|
|
||||||
preloadedTimes: []time.Time{
|
|
||||||
now,
|
|
||||||
now.Add(time.Minute),
|
|
||||||
now.Add(time.Minute * 2),
|
|
||||||
now.Add(time.Minute * 3),
|
|
||||||
now.Add(time.Minute * 4),
|
|
||||||
now.Add(time.Minute * 5),
|
|
||||||
now.Add(time.Minute * 6),
|
|
||||||
now.Add(time.Minute * 7),
|
|
||||||
now.Add(time.Minute * 8),
|
|
||||||
now.Add(time.Minute * 9),
|
|
||||||
},
|
|
||||||
expectedSchedule: []time.Time{
|
|
||||||
now,
|
|
||||||
now.Add(time.Second * 2),
|
|
||||||
now.Add(time.Minute * 1).Add(time.Second * 4),
|
|
||||||
now.Add(time.Minute * 2).Add(time.Second * 8),
|
|
||||||
now.Add(time.Minute * 3).Add(time.Second * 16),
|
|
||||||
now.Add(time.Minute * 4).Add(time.Second * 32),
|
|
||||||
now.Add(time.Minute * 5).Add(time.Second * 64),
|
|
||||||
now.Add(time.Minute * 6).Add(time.Minute * 2),
|
|
||||||
now.Add(time.Minute * 7).Add(time.Minute * 2),
|
|
||||||
now.Add(time.Minute * 8).Add(time.Minute * 2),
|
|
||||||
},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
for i, scenario := range scenarios {
|
|
||||||
provider := test.NewInstantProvider(scenario.preloadedTimes)
|
|
||||||
|
|
||||||
reporter := fakeHealthReporter{}
|
|
||||||
for _, state := range scenario.futureHealthState {
|
|
||||||
reporter.stateQueue = append(reporter.stateQueue, state)
|
|
||||||
}
|
|
||||||
if len(scenario.preloadedTimes) != len(scenario.futureHealthState) || len(scenario.futureHealthState) != len(scenario.expectedSchedule) {
|
|
||||||
t.Fatalf("%d. times and health reports and next time lengths were not equal.", i)
|
|
||||||
}
|
|
||||||
|
|
||||||
time := utility.Time{
|
|
||||||
Provider: provider,
|
|
||||||
}
|
|
||||||
|
|
||||||
scheduler := healthScheduler{
|
|
||||||
time: time,
|
|
||||||
target: reporter,
|
|
||||||
scheduledFor: now,
|
|
||||||
}
|
|
||||||
|
|
||||||
for j := 0; j < len(scenario.preloadedTimes); j++ {
|
|
||||||
futureState := scenario.futureHealthState[j]
|
|
||||||
scheduler.Reschedule(scenario.preloadedTimes[j], futureState)
|
|
||||||
nextSchedule := scheduler.ScheduledFor()
|
|
||||||
if nextSchedule != scenario.expectedSchedule[j] {
|
|
||||||
t.Errorf("%d.%d. Expected to be scheduled to %s, got %s", i, j, scenario.expectedSchedule[j], nextSchedule)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestHealthScheduler(t *testing.T) {
|
|
||||||
testHealthScheduler(t)
|
|
||||||
}
|
|
||||||
|
|
||||||
func BenchmarkHealthScheduler(b *testing.B) {
|
|
||||||
for i := 0; i < b.N; i++ {
|
|
||||||
testHealthScheduler(b)
|
|
||||||
}
|
|
||||||
}
|
|
|
@ -87,15 +87,6 @@ const (
|
||||||
UNREACHABLE
|
UNREACHABLE
|
||||||
)
|
)
|
||||||
|
|
||||||
// A healthReporter is a type that can provide insight into its health state.
|
|
||||||
//
|
|
||||||
// It mainly exists for testability reasons to decouple the scheduler behaviors
|
|
||||||
// from fully-fledged Target and other types.
|
|
||||||
type healthReporter interface {
|
|
||||||
// Report the last-known health state for this target.
|
|
||||||
State() TargetState
|
|
||||||
}
|
|
||||||
|
|
||||||
// A Target represents an endpoint that should be interrogated for metrics.
|
// A Target represents an endpoint that should be interrogated for metrics.
|
||||||
//
|
//
|
||||||
// The protocol described by this type will likely change in future iterations,
|
// The protocol described by this type will likely change in future iterations,
|
||||||
|
@ -108,26 +99,13 @@ type healthReporter interface {
|
||||||
// refers.
|
// refers.
|
||||||
type Target interface {
|
type Target interface {
|
||||||
// Retrieve values from this target.
|
// Retrieve values from this target.
|
||||||
//
|
Scrape(ingester extraction.Ingester) error
|
||||||
// earliest refers to the soonest available opportunity to reschedule the
|
|
||||||
// target for a future retrieval. It is up to the underlying scheduler type,
|
|
||||||
// alluded to in the scheduledFor function, to use this as it wants to. The
|
|
||||||
// current use case is to create a common batching time for scraping multiple
|
|
||||||
// Targets in the future through the TargetPool.
|
|
||||||
Scrape(earliest time.Time, ingester extraction.Ingester) error
|
|
||||||
// Fulfill the healthReporter interface.
|
|
||||||
State() TargetState
|
|
||||||
// Report the soonest time at which this Target may be scheduled for
|
|
||||||
// retrieval. This value needn't convey that the operation occurs at this
|
|
||||||
// time, but it should occur no sooner than it.
|
|
||||||
//
|
|
||||||
// Right now, this is used as the sorting key in TargetPool.
|
|
||||||
ScheduledFor() time.Time
|
|
||||||
// EstimatedTimeToExecute emits the amount of time until the next prospective
|
|
||||||
// scheduling opportunity for this target.
|
|
||||||
EstimatedTimeToExecute() time.Duration
|
|
||||||
// Return the last encountered scrape error, if any.
|
// Return the last encountered scrape error, if any.
|
||||||
LastError() error
|
LastError() error
|
||||||
|
// Return the health of the target.
|
||||||
|
State() TargetState
|
||||||
|
// Return the last time a scrape was attempted.
|
||||||
|
LastScrape() time.Time
|
||||||
// The address to which the Target corresponds. Out of all of the available
|
// The address to which the Target corresponds. Out of all of the available
|
||||||
// points in this interface, this one is the best candidate to change given
|
// points in this interface, this one is the best candidate to change given
|
||||||
// the ways to express the endpoint.
|
// the ways to express the endpoint.
|
||||||
|
@ -145,13 +123,12 @@ type Target interface {
|
||||||
|
|
||||||
// target is a Target that refers to a singular HTTP or HTTPS endpoint.
|
// target is a Target that refers to a singular HTTP or HTTPS endpoint.
|
||||||
type target struct {
|
type target struct {
|
||||||
// scheduler provides the scheduling strategy that is used to formulate what
|
|
||||||
// is returned in Target.scheduledFor.
|
|
||||||
scheduler scheduler
|
|
||||||
// The current health state of the target.
|
// The current health state of the target.
|
||||||
state TargetState
|
state TargetState
|
||||||
// The last encountered scrape error, if any.
|
// The last encountered scrape error, if any.
|
||||||
lastError error
|
lastError error
|
||||||
|
// The last time a scrape was attempted.
|
||||||
|
lastScrape time.Time
|
||||||
|
|
||||||
address string
|
address string
|
||||||
// What is the deadline for the HTTP or HTTPS against this endpoint.
|
// What is the deadline for the HTTP or HTTPS against this endpoint.
|
||||||
|
@ -171,11 +148,6 @@ func NewTarget(address string, deadline time.Duration, baseLabels clientmodel.La
|
||||||
httpClient: utility.NewDeadlineClient(deadline),
|
httpClient: utility.NewDeadlineClient(deadline),
|
||||||
}
|
}
|
||||||
|
|
||||||
scheduler := &healthScheduler{
|
|
||||||
target: target,
|
|
||||||
}
|
|
||||||
target.scheduler = scheduler
|
|
||||||
|
|
||||||
return target
|
return target
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -204,22 +176,18 @@ func (t *target) recordScrapeHealth(ingester extraction.Ingester, timestamp clie
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *target) Scrape(earliest time.Time, ingester extraction.Ingester) error {
|
func (t *target) Scrape(ingester extraction.Ingester) error {
|
||||||
now := clientmodel.Now()
|
now := clientmodel.Now()
|
||||||
futureState := t.state
|
|
||||||
err := t.scrape(now, ingester)
|
err := t.scrape(now, ingester)
|
||||||
if err != nil {
|
if err == nil {
|
||||||
t.recordScrapeHealth(ingester, now, false)
|
t.state = ALIVE
|
||||||
futureState = UNREACHABLE
|
|
||||||
} else {
|
|
||||||
t.recordScrapeHealth(ingester, now, true)
|
t.recordScrapeHealth(ingester, now, true)
|
||||||
futureState = ALIVE
|
} else {
|
||||||
|
t.state = UNREACHABLE
|
||||||
|
t.recordScrapeHealth(ingester, now, false)
|
||||||
}
|
}
|
||||||
|
t.lastScrape = time.Now()
|
||||||
t.scheduler.Reschedule(earliest, futureState)
|
|
||||||
t.state = futureState
|
|
||||||
t.lastError = err
|
t.lastError = err
|
||||||
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -275,20 +243,16 @@ func (t *target) scrape(timestamp clientmodel.Timestamp, ingester extraction.Ing
|
||||||
return processor.ProcessSingle(resp.Body, i, processOptions)
|
return processor.ProcessSingle(resp.Body, i, processOptions)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (t *target) LastError() error {
|
||||||
|
return t.lastError
|
||||||
|
}
|
||||||
|
|
||||||
func (t *target) State() TargetState {
|
func (t *target) State() TargetState {
|
||||||
return t.state
|
return t.state
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *target) ScheduledFor() time.Time {
|
func (t *target) LastScrape() time.Time {
|
||||||
return t.scheduler.ScheduledFor()
|
return t.lastScrape
|
||||||
}
|
|
||||||
|
|
||||||
func (t *target) EstimatedTimeToExecute() time.Duration {
|
|
||||||
return time.Now().Sub(t.scheduler.ScheduledFor())
|
|
||||||
}
|
|
||||||
|
|
||||||
func (t *target) LastError() error {
|
|
||||||
return t.lastError
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *target) Address() string {
|
func (t *target) Address() string {
|
||||||
|
@ -327,11 +291,3 @@ type targets []Target
|
||||||
func (t targets) Len() int {
|
func (t targets) Len() int {
|
||||||
return len(t)
|
return len(t)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t targets) Less(i, j int) bool {
|
|
||||||
return t[i].ScheduledFor().Before(t[j].ScheduledFor())
|
|
||||||
}
|
|
||||||
|
|
||||||
func (t targets) Swap(i, j int) {
|
|
||||||
t[i], t[j] = t[j], t[i]
|
|
||||||
}
|
|
||||||
|
|
|
@ -38,12 +38,11 @@ func (i *collectResultIngester) Ingest(r *extraction.Result) error {
|
||||||
|
|
||||||
func TestTargetScrapeUpdatesState(t *testing.T) {
|
func TestTargetScrapeUpdatesState(t *testing.T) {
|
||||||
testTarget := target{
|
testTarget := target{
|
||||||
scheduler: literalScheduler{},
|
|
||||||
state: UNKNOWN,
|
state: UNKNOWN,
|
||||||
address: "bad schema",
|
address: "bad schema",
|
||||||
httpClient: utility.NewDeadlineClient(0),
|
httpClient: utility.NewDeadlineClient(0),
|
||||||
}
|
}
|
||||||
testTarget.Scrape(time.Time{}, nopIngester{})
|
testTarget.Scrape(nopIngester{})
|
||||||
if testTarget.state != UNREACHABLE {
|
if testTarget.state != UNREACHABLE {
|
||||||
t.Errorf("Expected target state %v, actual: %v", UNREACHABLE, testTarget.state)
|
t.Errorf("Expected target state %v, actual: %v", UNREACHABLE, testTarget.state)
|
||||||
}
|
}
|
||||||
|
@ -51,7 +50,6 @@ func TestTargetScrapeUpdatesState(t *testing.T) {
|
||||||
|
|
||||||
func TestTargetRecordScrapeHealth(t *testing.T) {
|
func TestTargetRecordScrapeHealth(t *testing.T) {
|
||||||
testTarget := target{
|
testTarget := target{
|
||||||
scheduler: literalScheduler{},
|
|
||||||
address: "http://example.url",
|
address: "http://example.url",
|
||||||
baseLabels: clientmodel.LabelSet{clientmodel.JobLabel: "testjob"},
|
baseLabels: clientmodel.LabelSet{clientmodel.JobLabel: "testjob"},
|
||||||
httpClient: utility.NewDeadlineClient(0),
|
httpClient: utility.NewDeadlineClient(0),
|
||||||
|
@ -102,7 +100,7 @@ func TestTargetScrapeTimeout(t *testing.T) {
|
||||||
|
|
||||||
// scrape once without timeout
|
// scrape once without timeout
|
||||||
signal <- true
|
signal <- true
|
||||||
if err := testTarget.Scrape(time.Now(), ingester); err != nil {
|
if err := testTarget.Scrape(ingester); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -111,12 +109,12 @@ func TestTargetScrapeTimeout(t *testing.T) {
|
||||||
|
|
||||||
// now scrape again
|
// now scrape again
|
||||||
signal <- true
|
signal <- true
|
||||||
if err := testTarget.Scrape(time.Now(), ingester); err != nil {
|
if err := testTarget.Scrape(ingester); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// now timeout
|
// now timeout
|
||||||
if err := testTarget.Scrape(time.Now(), ingester); err == nil {
|
if err := testTarget.Scrape(ingester); err == nil {
|
||||||
t.Fatal("expected scrape to timeout")
|
t.Fatal("expected scrape to timeout")
|
||||||
} else {
|
} else {
|
||||||
signal <- true // let handler continue
|
signal <- true // let handler continue
|
||||||
|
@ -124,7 +122,7 @@ func TestTargetScrapeTimeout(t *testing.T) {
|
||||||
|
|
||||||
// now scrape again without timeout
|
// now scrape again without timeout
|
||||||
signal <- true
|
signal <- true
|
||||||
if err := testTarget.Scrape(time.Now(), ingester); err != nil {
|
if err := testTarget.Scrape(ingester); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -140,7 +138,7 @@ func TestTargetScrape404(t *testing.T) {
|
||||||
ingester := nopIngester{}
|
ingester := nopIngester{}
|
||||||
|
|
||||||
want := errors.New("server returned HTTP status 404 Not Found")
|
want := errors.New("server returned HTTP status 404 Not Found")
|
||||||
got := testTarget.Scrape(time.Now(), ingester)
|
got := testTarget.Scrape(ingester)
|
||||||
if got == nil || want.Error() != got.Error() {
|
if got == nil || want.Error() != got.Error() {
|
||||||
t.Fatalf("want err %q, got %q", want, got)
|
t.Fatalf("want err %q, got %q", want, got)
|
||||||
}
|
}
|
||||||
|
|
|
@ -55,7 +55,7 @@ func (t fakeTarget) Interval() time.Duration {
|
||||||
return t.interval
|
return t.interval
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *fakeTarget) Scrape(e time.Time, i extraction.Ingester) error {
|
func (t *fakeTarget) Scrape(i extraction.Ingester) error {
|
||||||
t.scrapeCount++
|
t.scrapeCount++
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
@ -65,6 +65,10 @@ func (t fakeTarget) State() TargetState {
|
||||||
return ALIVE
|
return ALIVE
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (t fakeTarget) LastScrape() time.Time {
|
||||||
|
return time.Now()
|
||||||
|
}
|
||||||
|
|
||||||
func (t *fakeTarget) ScheduledFor() (time time.Time) {
|
func (t *fakeTarget) ScheduledFor() (time time.Time) {
|
||||||
time = t.schedules[t.scheduleIndex]
|
time = t.schedules[t.scheduleIndex]
|
||||||
t.scheduleIndex++
|
t.scheduleIndex++
|
||||||
|
|
|
@ -14,7 +14,6 @@
|
||||||
package retrieval
|
package retrieval
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"sort"
|
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
@ -133,11 +132,11 @@ func (p *TargetPool) replaceTargets(newTargets []Target) {
|
||||||
p.targets = newTargets
|
p.targets = newTargets
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *TargetPool) runSingle(earliest time.Time, ingester extraction.Ingester, t Target) {
|
func (p *TargetPool) runSingle(ingester extraction.Ingester, t Target) {
|
||||||
p.manager.acquire()
|
p.manager.acquire()
|
||||||
defer p.manager.release()
|
defer p.manager.release()
|
||||||
|
|
||||||
t.Scrape(earliest, ingester)
|
t.Scrape(ingester)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *TargetPool) runIteration(ingester extraction.Ingester, interval time.Duration) {
|
func (p *TargetPool) runIteration(ingester extraction.Ingester, interval time.Duration) {
|
||||||
|
@ -156,23 +155,11 @@ func (p *TargetPool) runIteration(ingester extraction.Ingester, interval time.Du
|
||||||
begin := time.Now()
|
begin := time.Now()
|
||||||
wait := sync.WaitGroup{}
|
wait := sync.WaitGroup{}
|
||||||
|
|
||||||
// Sort p.targets by next scheduling time so we can process the earliest
|
|
||||||
// targets first.
|
|
||||||
sort.Sort(p.targets)
|
|
||||||
|
|
||||||
for _, target := range p.targets {
|
for _, target := range p.targets {
|
||||||
now := time.Now()
|
|
||||||
|
|
||||||
if target.ScheduledFor().After(now) {
|
|
||||||
// None of the remaining targets are ready to be scheduled. Signal that
|
|
||||||
// we're done processing them in this scrape iteration.
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
wait.Add(1)
|
wait.Add(1)
|
||||||
|
|
||||||
go func(t Target) {
|
go func(t Target) {
|
||||||
p.runSingle(now, ingester, t)
|
p.runSingle(ingester, t)
|
||||||
wait.Done()
|
wait.Done()
|
||||||
}(target)
|
}(target)
|
||||||
}
|
}
|
||||||
|
|
|
@ -14,7 +14,6 @@
|
||||||
package retrieval
|
package retrieval
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"sort"
|
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
@ -57,44 +56,21 @@ func testTargetPool(t testing.TB) {
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "plural descending schedules",
|
name: "plural schedules",
|
||||||
inputs: []input{
|
inputs: []input{
|
||||||
{
|
{
|
||||||
address: "http://plural-descending.com",
|
address: "http://plural.net",
|
||||||
scheduledFor: time.Date(2013, 1, 4, 12, 0, 0, 0, time.UTC),
|
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
address: "http://plural-descending.net",
|
address: "http://plural.com",
|
||||||
scheduledFor: time.Date(2013, 1, 4, 11, 0, 0, 0, time.UTC),
|
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
outputs: []output{
|
outputs: []output{
|
||||||
{
|
{
|
||||||
address: "http://plural-descending.net",
|
address: "http://plural.net",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
address: "http://plural-descending.com",
|
address: "http://plural.com",
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "plural ascending schedules",
|
|
||||||
inputs: []input{
|
|
||||||
{
|
|
||||||
address: "http://plural-ascending.net",
|
|
||||||
scheduledFor: time.Date(2013, 1, 4, 11, 0, 0, 0, time.UTC),
|
|
||||||
},
|
|
||||||
{
|
|
||||||
address: "http://plural-ascending.com",
|
|
||||||
scheduledFor: time.Date(2013, 1, 4, 12, 0, 0, 0, time.UTC),
|
|
||||||
},
|
|
||||||
},
|
|
||||||
outputs: []output{
|
|
||||||
{
|
|
||||||
address: "http://plural-ascending.net",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
address: "http://plural-ascending.com",
|
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
@ -105,13 +81,11 @@ func testTargetPool(t testing.TB) {
|
||||||
|
|
||||||
for _, input := range scenario.inputs {
|
for _, input := range scenario.inputs {
|
||||||
target := target{
|
target := target{
|
||||||
address: input.address,
|
address: input.address,
|
||||||
scheduler: literalScheduler(input.scheduledFor),
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pool.addTarget(&target)
|
pool.addTarget(&target)
|
||||||
}
|
}
|
||||||
sort.Sort(pool.targets)
|
|
||||||
|
|
||||||
if pool.targets.Len() != len(scenario.outputs) {
|
if pool.targets.Len() != len(scenario.outputs) {
|
||||||
t.Errorf("%s %d. expected TargetPool size to be %d but was %d", scenario.name, i, len(scenario.outputs), pool.targets.Len())
|
t.Errorf("%s %d. expected TargetPool size to be %d but was %d", scenario.name, i, len(scenario.outputs), pool.targets.Len())
|
||||||
|
@ -136,68 +110,41 @@ func TestTargetPool(t *testing.T) {
|
||||||
testTargetPool(t)
|
testTargetPool(t)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestTargetPoolIterationWithUnhealthyTargetsFinishes(t *testing.T) {
|
|
||||||
pool := TargetPool{}
|
|
||||||
target := &target{
|
|
||||||
address: "http://example.com/metrics.json",
|
|
||||||
scheduler: literalScheduler(time.Date(9999, 1, 1, 0, 0, 0, 0, time.UTC)),
|
|
||||||
}
|
|
||||||
pool.addTarget(target)
|
|
||||||
|
|
||||||
done := make(chan bool)
|
|
||||||
go func() {
|
|
||||||
pool.runIteration(nopIngester{}, time.Duration(0))
|
|
||||||
done <- true
|
|
||||||
}()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case <-done:
|
|
||||||
break
|
|
||||||
case <-time.After(time.Duration(1) * time.Second):
|
|
||||||
t.Fatalf("Targetpool iteration is stuck")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestTargetPoolReplaceTargets(t *testing.T) {
|
func TestTargetPoolReplaceTargets(t *testing.T) {
|
||||||
pool := TargetPool{}
|
pool := TargetPool{}
|
||||||
oldTarget1 := &target{
|
oldTarget1 := &target{
|
||||||
address: "http://example1.com/metrics.json",
|
address: "http://example1.com/metrics.json",
|
||||||
scheduler: literalScheduler(time.Date(9999, 1, 1, 0, 0, 0, 0, time.UTC)),
|
state: UNREACHABLE,
|
||||||
state: UNREACHABLE,
|
|
||||||
}
|
}
|
||||||
oldTarget2 := &target{
|
oldTarget2 := &target{
|
||||||
address: "http://example2.com/metrics.json",
|
address: "http://example2.com/metrics.json",
|
||||||
scheduler: literalScheduler(time.Date(7500, 1, 1, 0, 0, 0, 0, time.UTC)),
|
state: UNREACHABLE,
|
||||||
state: UNREACHABLE,
|
|
||||||
}
|
}
|
||||||
newTarget1 := &target{
|
newTarget1 := &target{
|
||||||
address: "http://example1.com/metrics.json",
|
address: "http://example1.com/metrics.json",
|
||||||
scheduler: literalScheduler(time.Date(5000, 1, 1, 0, 0, 0, 0, time.UTC)),
|
state: ALIVE,
|
||||||
state: ALIVE,
|
|
||||||
}
|
}
|
||||||
newTarget2 := &target{
|
newTarget2 := &target{
|
||||||
address: "http://example3.com/metrics.json",
|
address: "http://example3.com/metrics.json",
|
||||||
scheduler: literalScheduler(time.Date(2500, 1, 1, 0, 0, 0, 0, time.UTC)),
|
state: ALIVE,
|
||||||
state: ALIVE,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pool.addTarget(oldTarget1)
|
pool.addTarget(oldTarget1)
|
||||||
pool.addTarget(oldTarget2)
|
pool.addTarget(oldTarget2)
|
||||||
|
|
||||||
pool.replaceTargets([]Target{newTarget1, newTarget2})
|
pool.replaceTargets([]Target{newTarget1, newTarget2})
|
||||||
sort.Sort(pool.targets)
|
|
||||||
|
|
||||||
if pool.targets.Len() != 2 {
|
if pool.targets.Len() != 2 {
|
||||||
t.Errorf("Expected 2 elements in pool, had %d", pool.targets.Len())
|
t.Errorf("Expected 2 elements in pool, had %d", pool.targets.Len())
|
||||||
}
|
}
|
||||||
|
|
||||||
target1 := pool.targets[0].(*target)
|
target1 := pool.targets[0].(*target)
|
||||||
if target1.state != newTarget1.state {
|
if target1.state != oldTarget1.state {
|
||||||
t.Errorf("Wrong first target returned from pool, expected %v, got %v", newTarget2, target1)
|
t.Errorf("Wrong first target returned from pool, expected %v, got %v", oldTarget1, target1)
|
||||||
}
|
}
|
||||||
target2 := pool.targets[1].(*target)
|
target2 := pool.targets[1].(*target)
|
||||||
if target2.state != oldTarget1.state {
|
if target2.state != newTarget2.state {
|
||||||
t.Errorf("Wrong second target returned from pool, expected %v, got %v", oldTarget1, target2)
|
t.Errorf("Wrong second target returned from pool, expected %v, got %v", newTarget2, target2)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -260,14 +260,14 @@ func (te templateExpander) ExpandHTML(templateFiles []string) (result string, re
|
||||||
|
|
||||||
var buffer bytes.Buffer
|
var buffer bytes.Buffer
|
||||||
tmpl := html_template.New(te.name).Funcs(html_template.FuncMap(te.funcMap))
|
tmpl := html_template.New(te.name).Funcs(html_template.FuncMap(te.funcMap))
|
||||||
tmpl.Funcs(html_template.FuncMap{
|
tmpl.Funcs(html_template.FuncMap{
|
||||||
"tmpl": func(name string, data interface{}) (html_template.HTML, error) {
|
"tmpl": func(name string, data interface{}) (html_template.HTML, error) {
|
||||||
var buffer bytes.Buffer
|
var buffer bytes.Buffer
|
||||||
err := tmpl.ExecuteTemplate(&buffer, name, data)
|
err := tmpl.ExecuteTemplate(&buffer, name, data)
|
||||||
return html_template.HTML(buffer.String()), err
|
return html_template.HTML(buffer.String()), err
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
tmpl, err := tmpl.Parse(te.text)
|
tmpl, err := tmpl.Parse(te.text)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", fmt.Errorf("Error parsing template %v: %v", te.name, err)
|
return "", fmt.Errorf("Error parsing template %v: %v", te.name, err)
|
||||||
}
|
}
|
||||||
|
|
|
@ -39,7 +39,7 @@
|
||||||
<th>Endpoint</th>
|
<th>Endpoint</th>
|
||||||
<th>State</th>
|
<th>State</th>
|
||||||
<th>Base Labels</th>
|
<th>Base Labels</th>
|
||||||
<th>Next Retrieval</th>
|
<th>Last Scrape</th>
|
||||||
<th>Error</th>
|
<th>Error</th>
|
||||||
</tr>
|
</tr>
|
||||||
</thead>
|
</thead>
|
||||||
|
@ -56,7 +56,7 @@
|
||||||
{{.BaseLabels}}
|
{{.BaseLabels}}
|
||||||
</td>
|
</td>
|
||||||
<td>
|
<td>
|
||||||
{{.EstimatedTimeToExecute}}
|
{{if .LastScrape.IsZero}}Never{{else}}{{since .LastScrape}} ago{{end}}
|
||||||
</td>
|
</td>
|
||||||
<td>
|
<td>
|
||||||
{{if .LastError}}
|
{{if .LastError}}
|
||||||
|
|
55
web/web.go
55
web/web.go
|
@ -17,6 +17,7 @@ import (
|
||||||
"flag"
|
"flag"
|
||||||
"fmt"
|
"fmt"
|
||||||
"html/template"
|
"html/template"
|
||||||
|
"io/ioutil"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
_ "net/http/pprof"
|
_ "net/http/pprof"
|
||||||
|
@ -113,44 +114,42 @@ func (s WebService) quitHandler(w http.ResponseWriter, r *http.Request) {
|
||||||
s.QuitDelegate()
|
s.QuitDelegate()
|
||||||
}
|
}
|
||||||
|
|
||||||
func getLocalTemplate(name string) (*template.Template, error) {
|
func getTemplateFile(name string) (string, error) {
|
||||||
return template.ParseFiles(
|
if *useLocalAssets {
|
||||||
"web/templates/_base.html",
|
file, err := ioutil.ReadFile(fmt.Sprintf("web/templates/%s.html", name))
|
||||||
fmt.Sprintf("web/templates/%s.html", name),
|
if err != nil {
|
||||||
)
|
glog.Errorf("Could not read %s template: %s", name, err)
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return string(file), nil
|
||||||
|
} else {
|
||||||
|
file, err := blob.GetFile(blob.TemplateFiles, name+".html")
|
||||||
|
if err != nil {
|
||||||
|
glog.Errorf("Could not read %s template: %s", name, err)
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return string(file), nil
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func getEmbeddedTemplate(name string) (*template.Template, error) {
|
func getTemplate(name string) (t *template.Template, err error) {
|
||||||
t := template.New("_base")
|
t = template.New("_base")
|
||||||
|
t.Funcs(template.FuncMap{
|
||||||
file, err := blob.GetFile(blob.TemplateFiles, "_base.html")
|
"since": time.Since,
|
||||||
|
})
|
||||||
|
file, err := getTemplateFile("_base")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
glog.Error("Could not read base template: ", err)
|
glog.Error("Could not read base template: ", err)
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
t.Parse(string(file))
|
t.Parse(file)
|
||||||
|
|
||||||
file, err = blob.GetFile(blob.TemplateFiles, name+".html")
|
file, err = getTemplateFile(name)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
glog.Errorf("Could not read %s template: %s", name, err)
|
glog.Error("Could not read base template: ", err)
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
t.Parse(string(file))
|
t.Parse(file)
|
||||||
|
|
||||||
return t, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func getTemplate(name string) (t *template.Template, err error) {
|
|
||||||
if *useLocalAssets {
|
|
||||||
t, err = getLocalTemplate(name)
|
|
||||||
} else {
|
|
||||||
t, err = getEmbeddedTemplate(name)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue