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

Extract appending from goroutine.

pull/84/head
Matt T. Proud 2013-03-14 18:09:19 -07:00
parent 187cd4cdbc
commit 47ce7ad302
1 changed files with 37 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
storage/metric/leveldb.go
View File

@ -540,7 +540,6 @@ func (l *LevelDBMetricPersistence) AppendSamples(samples model.Samples) (err err
var ( var (
fingerprintToSamples = groupByFingerprint(samples) fingerprintToSamples = groupByFingerprint(samples)
indexErrChan = make(chan error) indexErrChan = make(chan error)
doneCommitting sync.WaitGroup
) )
go func(groups map[model.Fingerprint]model.Samples) { go func(groups map[model.Fingerprint]model.Samples) {
@ -555,55 +554,50 @@ func (l *LevelDBMetricPersistence) AppendSamples(samples model.Samples) (err err
indexErrChan <- l.indexMetrics(metrics) indexErrChan <- l.indexMetrics(metrics)
}(fingerprintToSamples) }(fingerprintToSamples)
go func() { samplesBatch := leveldb.NewBatch()
doneCommitting.Add(1) defer samplesBatch.Close()
samplesBatch := leveldb.NewBatch()
defer samplesBatch.Close()
defer doneCommitting.Done()
for fingerprint, group := range fingerprintToSamples { for fingerprint, group := range fingerprintToSamples {
for { for {
lengthOfGroup := len(group) lengthOfGroup := len(group)
if lengthOfGroup == 0 { if lengthOfGroup == 0 {
break break
}
take := *leveldbChunkSize
if lengthOfGroup < take {
take = lengthOfGroup
}
chunk := group[0:take]
group = group[take:lengthOfGroup]
key := &dto.SampleKey{
Fingerprint: fingerprint.ToDTO(),
Timestamp: indexable.EncodeTime(chunk[0].Timestamp),
LastTimestamp: proto.Int64(chunk[take-1].Timestamp.Unix()),
SampleCount: proto.Uint32(uint32(take)),
}
value := &dto.SampleValueSeries{}
for _, sample := range chunk {
value.Value = append(value.Value, &dto.SampleValueSeries_Value{
Timestamp: proto.Int64(sample.Timestamp.Unix()),
Value: proto.Float32(float32(sample.Value)),
})
}
samplesBatch.Put(coding.NewProtocolBufferEncoder(key), coding.NewProtocolBufferEncoder(value))
} }
take := *leveldbChunkSize
if lengthOfGroup < take {
take = lengthOfGroup
}
chunk := group[0:take]
group = group[take:lengthOfGroup]
key := &dto.SampleKey{
Fingerprint: fingerprint.ToDTO(),
Timestamp: indexable.EncodeTime(chunk[0].Timestamp),
LastTimestamp: proto.Int64(chunk[take-1].Timestamp.Unix()),
SampleCount: proto.Uint32(uint32(take)),
}
value := &dto.SampleValueSeries{}
for _, sample := range chunk {
value.Value = append(value.Value, &dto.SampleValueSeries_Value{
Timestamp: proto.Int64(sample.Timestamp.Unix()),
Value: proto.Float32(float32(sample.Value)),
})
}
samplesBatch.Put(coding.NewProtocolBufferEncoder(key), coding.NewProtocolBufferEncoder(value))
} }
}
err = l.metricSamples.Commit(samplesBatch) err = l.metricSamples.Commit(samplesBatch)
if err != nil { if err != nil {
panic(err) panic(err)
} }
}()
doneCommitting.Wait()
err = <-indexErrChan err = <-indexErrChan
if err != nil { if err != nil {
panic(err) panic(err)