github
/
prometheus
mirror of https://github.com/prometheus/prometheus
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Report correct number of appended samples

pull/5805/head
Fabian Reinartz 2017-01-10 11:17:37 +01:00
parent 29883a18fc
commit d86e8a63c7
2 changed files with 20 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
cmd/tsdb/main.go
View File

@ -121,7 +121,7 @@ func (b *writeBenchmark) run(cmd *cobra.Command, args []string) {
measureTime("ingestScrapes", func() { measureTime("ingestScrapes", func() {
b.startProfiling() b.startProfiling()
if err := b.ingestScrapes(metrics, 3000); err != nil { if err := b.ingestScrapes(metrics, 2000); err != nil {
exitWithError(err) exitWithError(err)
} }
}) })
@ -135,6 +135,8 @@ func (b *writeBenchmark) run(cmd *cobra.Command, args []string) {
func (b *writeBenchmark) ingestScrapes(metrics []model.Metric, scrapeCount int) error { func (b *writeBenchmark) ingestScrapes(metrics []model.Metric, scrapeCount int) error {
var wg sync.WaitGroup var wg sync.WaitGroup
var mu sync.Mutex
var total uint64
for len(metrics) > 0 { for len(metrics) > 0 {
l := 1000 l := 1000
@ -145,20 +147,25 @@ func (b *writeBenchmark) ingestScrapes(metrics []model.Metric, scrapeCount int)
metrics = metrics[l:] metrics = metrics[l:]
wg.Add(1) wg.Add(1)
go func() { go func(batch []model.Metric) {
if err := b.ingestScrapesShard(batch, scrapeCount); err != nil { n, err := b.ingestScrapesShard(batch, scrapeCount)
if err != nil {
// exitWithError(err) // exitWithError(err)
fmt.Println(" err", err) fmt.Println(" err", err)
} }
mu.Lock()
total += n
mu.Unlock()
wg.Done() wg.Done()
}() }(batch)
} }
wg.Wait() wg.Wait()
fmt.Println("> total samples:", total)
return nil return nil
} }
func (b *writeBenchmark) ingestScrapesShard(metrics []model.Metric, scrapeCount int) error { func (b *writeBenchmark) ingestScrapesShard(metrics []model.Metric, scrapeCount int) (uint64, error) {
app := b.storage.Appender() app := b.storage.Appender()
ts := int64(model.Now()) ts := int64(model.Now())
@ -181,6 +188,7 @@ func (b *writeBenchmark) ingestScrapesShard(metrics []model.Metric, scrapeCount
value: 123456789, value: 123456789,
} }
} }
total := uint64(0)
for i := 0; i < scrapeCount; i++ { for i := 0; i < scrapeCount; i++ {
ts += int64(30000) ts += int64(30000)
@ -188,12 +196,13 @@ func (b *writeBenchmark) ingestScrapesShard(metrics []model.Metric, scrapeCount
for _, s := range scrape { for _, s := range scrape {
s.value += 1000 s.value += 1000
app.Add(s.labels, ts, float64(s.value)) app.Add(s.labels, ts, float64(s.value))
total++
} }
if err := app.Commit(); err != nil { if err := app.Commit(); err != nil {
return err return total, err
} }
} }
return nil return total, nil
} }
func (b *writeBenchmark) startProfiling() { func (b *writeBenchmark) startProfiling() {

14
head.go
View File

@ -143,17 +143,9 @@ func (c *safeChunk) Iterator() chunks.Iterator {
return c.cd.iterator() return c.cd.iterator()
} }
func (c *safeChunk) Appender() (chunks.Appender, error) { func (c *safeChunk) Appender() (chunks.Appender, error) { panic("illegal") }
panic("illegal") func (c *safeChunk) Bytes() []byte { panic("illegal") }
} func (c *safeChunk) Encoding() chunks.Encoding { panic("illegal") }
func (c *safeChunk) Bytes() []byte {
panic("illegal")
}
func (c *safeChunk) Encoding() chunks.Encoding {
panic("illegal")
}
func (h *HeadBlock) interval() (int64, int64) { func (h *HeadBlock) interval() (int64, int64) {
h.bstats.mtx.RLock() h.bstats.mtx.RLock()