Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Reuse the same appender for report and scrape #7562

Merged
merged 13 commits into from
Jul 16, 2020
Merged

Reuse the same appender for report and scrape #7562

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
ee7250b
Scrape tests: implement isolation in collectResultAppender
roidelapluie Jul 11, 2020
05c29d0
Reuse the same appender for report and scrape
roidelapluie Jul 12, 2020
caa9fbc
Add test
roidelapluie Jul 12, 2020
d8bdcf1
Remove un-needed function call
roidelapluie Jul 13, 2020
fbc0871
Rename function
roidelapluie Jul 13, 2020
4b70df9
Move the function around and add a comment
roidelapluie Jul 13, 2020
c8f6e72
log errors
roidelapluie Jul 13, 2020
ec9b83a
Apply comments
roidelapluie Jul 14, 2020
c6a104d
Rework commit/rollback
roidelapluie Jul 14, 2020
41e3ac8
Fix logic
roidelapluie Jul 14, 2020
844fb36
No new alloc
roidelapluie Jul 14, 2020
4204288
reword
roidelapluie Jul 15, 2020
fd8f29f
Rename test
roidelapluie Jul 16, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 23 additions & 6 deletions scrape/helpers_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,8 +40,10 @@ type sample struct {
// collectResultAppender records all samples that were added through the appender.
// It can be used as its zero value or be backed by another appender it writes samples through.
type collectResultAppender struct {
next storage.Appender
result []sample
next storage.Appender
result []sample
pendingResult []sample
rolledbackResult []sample

mapper map[uint64]labels.Labels
}
Expand All @@ -55,7 +57,7 @@ func (a *collectResultAppender) AddFast(ref uint64, t int64, v float64) error {
if err != nil {
return err
}
a.result = append(a.result, sample{
a.pendingResult = append(a.pendingResult, sample{
metric: a.mapper[ref],
t: t,
v: v,
Expand All @@ -64,7 +66,7 @@ func (a *collectResultAppender) AddFast(ref uint64, t int64, v float64) error {
}

func (a *collectResultAppender) Add(m labels.Labels, t int64, v float64) (uint64, error) {
a.result = append(a.result, sample{
a.pendingResult = append(a.pendingResult, sample{
metric: m,
t: t,
v: v,
Expand All @@ -85,5 +87,20 @@ func (a *collectResultAppender) Add(m labels.Labels, t int64, v float64) (uint64
return ref, nil
}

func (a *collectResultAppender) Commit() error { return nil }
func (a *collectResultAppender) Rollback() error { return nil }
func (a *collectResultAppender) Commit() error {
a.result = append(a.result, a.pendingResult...)
a.pendingResult = nil
if a.next == nil {
return nil
}
return a.next.Commit()
}

func (a *collectResultAppender) Rollback() error {
a.rolledbackResult = a.pendingResult
a.pendingResult = nil
if a.next == nil {
return nil
}
return a.next.Rollback()
}
179 changes: 97 additions & 82 deletions scrape/scrape.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -932,77 +932,99 @@ mainLoop:
default:
}

var (
start = time.Now()
scrapeCtx, cancel = context.WithTimeout(sl.ctx, timeout)
)
last = sl.scrapeAndReport(interval, timeout, last, errc)

// Only record after the first scrape.
if !last.IsZero() {
targetIntervalLength.WithLabelValues(interval.String()).Observe(
time.Since(last).Seconds(),
)
select {
case <-sl.parentCtx.Done():
close(sl.stopped)
return
case <-sl.ctx.Done():
break mainLoop
case <-ticker.C:
}
}

b := sl.buffers.Get(sl.lastScrapeSize).([]byte)
buf := bytes.NewBuffer(b)
close(sl.stopped)

contentType, scrapeErr := sl.scraper.scrape(scrapeCtx, buf)
cancel()
if !sl.disabledEndOfRunStalenessMarkers {
sl.endOfRunStaleness(last, ticker, interval)
}
}

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)
}
} else {
level.Debug(sl.l).Log("msg", "Scrape failed", "err", scrapeErr.Error())
if errc != nil {
errc <- scrapeErr
}
}
// scrapeAndReport performs a scrape and then appends the result to the storage
// together with reporting metrics, all using the same appender.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"the same" is a little misleading as a 2nd appender will be created if there's a scrape error.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But in case of a scrape error, the 1st appender is rolled back, so it will not append anything to the storage, so only one and the same appender will ever append any samples to the storage.

But if you have a better suggestion for the wording, I'm sure @roidelapluie will appreciate it.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

// 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.

func (sl *scrapeLoop) scrapeAndReport(interval, timeout time.Duration, last time.Time, errc chan<- error) time.Time {
var (
start = time.Now()
scrapeCtx, cancel = context.WithTimeout(sl.ctx, timeout)
)

// 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(b, contentType, start)
if appErr != nil {
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([]byte{}, "", start); err != nil {
level.Warn(sl.l).Log("msg", "Append failed", "err", err)
}
}
// Only record after the first scrape.
if !last.IsZero() {
targetIntervalLength.WithLabelValues(interval.String()).Observe(
time.Since(last).Seconds(),
)
}

sl.buffers.Put(b)
b := sl.buffers.Get(sl.lastScrapeSize).([]byte)
buf := bytes.NewBuffer(b)

if scrapeErr == nil {
scrapeErr = appErr
}
contentType, scrapeErr := sl.scraper.scrape(scrapeCtx, buf)
cancel()

if err := sl.report(start, time.Since(start), total, added, seriesAdded, scrapeErr); err != nil {
level.Warn(sl.l).Log("msg", "Appending scrape report failed", "err", err)
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)
}
} else {
level.Debug(sl.l).Log("msg", "Scrape failed", "err", scrapeErr.Error())
if errc != nil {
errc <- scrapeErr
}
last = start
}

select {
case <-sl.parentCtx.Done():
close(sl.stopped)
app := sl.appender()
var err error
defer func() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's no early returns, so I don't see a need to use a defer here compared to putting this code at the end of the function - which would be clearer.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We guard ourselves against potential panic()

if err != nil {
app.Rollback()
return
case <-sl.ctx.Done():
break mainLoop
case <-ticker.C:
}
err = app.Commit()
if err != nil {
level.Error(sl.l).Log("msg", "Scrape commit failed", "err", err)
}
}()
// 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, start)
if appErr != nil {
app.Rollback()
app = sl.appender()
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{}, "", start); err != nil {
app.Rollback()
app = sl.appender()
level.Warn(sl.l).Log("msg", "Append failed", "err", err)
}
}

close(sl.stopped)
sl.buffers.Put(b)

if !sl.disabledEndOfRunStalenessMarkers {
sl.endOfRunStaleness(last, ticker, interval)
if scrapeErr == nil {
scrapeErr = appErr
}

if err = sl.report(app, start, time.Since(start), total, added, seriesAdded, scrapeErr); err != nil {
level.Warn(sl.l).Log("msg", "Appending scrape report failed", "err", err)
}
return start
}

func (sl *scrapeLoop) endOfRunStaleness(last time.Time, ticker *time.Ticker, interval time.Duration) {
Expand Down Expand Up @@ -1045,11 +1067,25 @@ func (sl *scrapeLoop) endOfRunStaleness(last time.Time, ticker *time.Ticker, int
// 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.
if _, _, _, err := sl.append([]byte{}, "", staleTime); err != nil {
level.Error(sl.l).Log("msg", "stale append failed", "err", err)
app := sl.appender()
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()
level.Warn(sl.l).Log("msg", "Stale append failed", "err", err)
}
if err := sl.reportStale(staleTime); err != nil {
level.Error(sl.l).Log("msg", "stale report failed", "err", err)
if err = sl.reportStale(app, staleTime); err != nil {
level.Warn(sl.l).Log("msg", "Stale report failed", "err", err)
}
}

Expand All @@ -1074,9 +1110,8 @@ type appendErrors struct {
numOutOfBounds int
}

func (sl *scrapeLoop) append(b []byte, contentType string, ts time.Time) (total, added, seriesAdded int, err error) {
func (sl *scrapeLoop) append(app storage.Appender, b []byte, contentType string, ts time.Time) (total, added, seriesAdded int, err error) {
var (
app = sl.appender()
p = textparse.New(b, contentType)
defTime = timestamp.FromTime(ts)
appErrs = appendErrors{}
Expand All @@ -1085,10 +1120,6 @@ func (sl *scrapeLoop) append(b []byte, contentType string, ts time.Time) (total,

defer func() {
if err != nil {
app.Rollback()
return
}
if err = app.Commit(); err != nil {
return
}
// Only perform cache cleaning if the scrape was not empty.
Expand Down Expand Up @@ -1275,7 +1306,7 @@ const (
scrapeSeriesAddedMetricName = "scrape_series_added" + "\xff"
)

func (sl *scrapeLoop) report(start time.Time, duration time.Duration, scraped, added, seriesAdded int, scrapeErr error) (err error) {
func (sl *scrapeLoop) report(app storage.Appender, start time.Time, duration time.Duration, scraped, added, seriesAdded int, scrapeErr error) (err error) {
sl.scraper.Report(start, duration, scrapeErr)

ts := timestamp.FromTime(start)
Expand All @@ -1284,14 +1315,6 @@ func (sl *scrapeLoop) report(start time.Time, duration time.Duration, scraped, a
if scrapeErr == nil {
health = 1
}
app := sl.appender()
defer func() {
if err != nil {
app.Rollback()
return
}
err = app.Commit()
}()

if err = sl.addReportSample(app, scrapeHealthMetricName, ts, health); err != nil {
return
Expand All @@ -1311,16 +1334,8 @@ func (sl *scrapeLoop) report(start time.Time, duration time.Duration, scraped, a
return
}

func (sl *scrapeLoop) reportStale(start time.Time) (err error) {
func (sl *scrapeLoop) reportStale(app storage.Appender, start time.Time) (err error) {
ts := timestamp.FromTime(start)
app := sl.appender()
defer func() {
if err != nil {
app.Rollback()
return
}
err = app.Commit()
}()

stale := math.Float64frombits(value.StaleNaN)

Expand Down
Loading