Aggr dump

.gitignore

@@ -1,5 +1,6 @@
 logpeckd
 logmocker
 logpeckd.log
+logpeck.db
 .test.log
 .unittest.db

aggregator.go

@@ -2,7 +2,7 @@ package logpeck
 
 import (
 	"strconv"
-	"time"
+	"sync"
 
 	log "github.com/Sirupsen/logrus"
 )
@@ -26,47 +26,33 @@ type AggregatorOption struct {
 
 // Aggregator .
 type Aggregator struct {
-	config   AggregatorConfig
-	buckets  map[string]map[string][]float64
-	postTime int64
+	config  AggregatorConfig
+	buckets map[string]map[string][]float64
+	mu      sync.Mutex
 }
 
 // NewAggregator create aggregator
 func NewAggregator(config *AggregatorConfig) *Aggregator {
 	aggregator := &Aggregator{
-		config:   *config,
-		buckets:  make(map[string]map[string][]float64),
-		postTime: 0,
+		config:  *config,
+		buckets: make(map[string]map[string][]float64),
 	}
 	return aggregator
 }
 
-func getSampleTime(ts int64, interval int64) int64 {
-	return ts / interval
-}
-
 // IsEnable return true if enable
 func (p *Aggregator) IsEnable() bool {
 	return p.config.Enable
 }
 
-// IsDeadline return true if reaching config.Interval
-func (p *Aggregator) IsDeadline(timestamp int64) bool {
-	interval := p.config.Interval
-	nowTime := getSampleTime(timestamp, interval)
-	if p.postTime != nowTime {
-		return true
-	}
-	return false
-}
-
 // Record fields
-func (p *Aggregator) Record(fields map[string]interface{}) int64 {
-	var now int64
+func (p *Aggregator) Record(fields map[string]interface{}) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	for i := 0; i < len(p.config.Options); i++ {
 		tags := p.config.Options[i].Tags
 		target := p.config.Options[i].Target
-		timestamp := p.config.Options[i].Timestamp
+		//timestamp := p.config.Options[i].Timestamp
 		bucketName := p.config.Options[i].PreMeasurment + "_" + p.config.Options[i].Measurment + "_" + target
 		bucketTag := ""
 		if p.config.Options[i].PreMeasurment != "" {
@@ -78,28 +64,26 @@ func (p *Aggregator) Record(fields map[string]interface{}) int64 {
 			measurment, ok := fields[p.config.Options[i].Measurment].(string)
 			if !ok {
 				log.Debug("[Record] Fields[measurment] format error: Fields[measurment] must be a string")
-				now = time.Now().Unix()
 				continue
 			}
 			bucketTag += measurment + "_" + target
 		}
 
-		//get time
-		var err error
-		timestampTmp, ok := fields[timestamp].(string)
-		if !ok {
-			now = time.Now().Unix()
-		} else {
-			now, err = strconv.ParseInt(timestampTmp, 10, 64)
-			if err != nil {
-				log.Debugf("[Record] timestamp:%v can't use strconv.ParseInt", timestampTmp)
-				now = time.Now().Unix()
+		/*
+			//get time
+			var err error
+			timestampTmp, ok := fields[timestamp].(string)
+			if ok {
+				ts, err := strconv.ParseInt(timestampTmp, 10, 64)
+				if err == nil {
+					log.Debugf("[Record] timestamp:%v can't use strconv.ParseInt", timestampTmp)
+				}
 			}
-		}
+		*/
 
 		if target == "" {
 			log.Debug("[Record] Target is error: Target is null")
-			return time.Now().Unix()
+			return
 		}
 		for i := 0; i < len(tags); i++ {
 			tagsTmp, ok := fields[tags[i]].(string)
@@ -113,7 +97,7 @@ func (p *Aggregator) Record(fields map[string]interface{}) int64 {
 		aggValue, ok := fields[target].(string)
 		if !ok {
 			log.Debugf("[Record] Fields[aggValue] format error: %v", fields[target])
-			return now
+			return
 		}
 		if _, ok := p.buckets[bucketName]; !ok {
 			p.buckets[bucketName] = make(map[string][]float64)
@@ -126,7 +110,7 @@ func (p *Aggregator) Record(fields map[string]interface{}) int64 {
 			p.buckets[bucketName][bucketTag] = append(p.buckets[bucketName][bucketTag], aggValueFloat64)
 		}
 	}
-	return now
+	return
 }
 
 func quickSort(values []float64, left, right int64) {
@@ -212,7 +196,9 @@ func getAggregation(targetValue []float64, aggregations []string) map[string]flo
 }
 
 // Dump aggregation result
-func (p *Aggregator) Dump(timestamp int64) map[string]interface{} {
+func (p *Aggregator) Dump() map[string]interface{} {
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	fields := map[string]interface{}{}
 	log.Debug("[Dump] bucket is", p.buckets)
 	//now := strconv.FormatInt(timestamp, 10)
@@ -228,8 +214,6 @@ func (p *Aggregator) Dump(timestamp int64) map[string]interface{} {
 			fields[bucketTag] = getAggregation(targetValue, aggregations)
 		}
 	}
-	fields["timestamp"] = timestamp
-	p.postTime = getSampleTime(timestamp, p.config.Interval)
 	p.buckets = map[string]map[string][]float64{}
 	log.Debug("[Dump] fields is", fields)
 	return fields

aggregator_test.go

@@ -1,11 +1,13 @@
 package logpeck
 
 import (
-	log "github.com/Sirupsen/logrus"
 	"strconv"
 	"testing"
+
+	log "github.com/Sirupsen/logrus"
 )
 
+/*
 func TestStartSend(*testing.T) {
 	log.Infof("[aggregator_test] TestStartSend")
 
@@ -25,16 +27,9 @@ func TestStartSend(*testing.T) {
 		Options:  options,
 	}
 	aggregator := NewAggregator(&aggregatorConfig)
-
-	deadline := aggregator.IsDeadline(int64(29))
-	if deadline == true {
-		panic(aggregator)
-	}
-	deadline = aggregator.IsDeadline(int64(31))
-	if deadline == false {
-		panic(aggregator)
-	}
+	aggregator.recordTime = 29
 }
+*/
 
 func TestRecord(*testing.T) {
 	test := AggregatorOption{
@@ -59,15 +54,11 @@ func TestRecord(*testing.T) {
 	fields["upstream"] = "127.0.0.1"
 	fields["cost"] = "2"
 	fields["time"] = "15"
-	if aggregator.Record(fields) != int64(15) {
-		panic(fields)
-	}
+	aggregator.Record(fields)
 	if aggregator.buckets["Test_aaa_cost"]["Test_getTest_cost,upstream=127.0.0.1"][0] != 2 {
 		panic(aggregator)
 	}
-	if aggregator.Record(fields) != int64(15) {
-		panic(fields)
-	}
+	aggregator.Record(fields)
 	if aggregator.buckets["Test_aaa_cost"]["Test_getTest_cost,upstream=127.0.0.1"][0]+aggregator.buckets["Test_aaa_cost"]["Test_getTest_cost,upstream=127.0.0.1"][1] != 4 {
 		panic(aggregator)
 	}
@@ -100,7 +91,7 @@ func TestDump(*testing.T) {
 		fields["cost"] = strconv.Itoa(i)
 		aggregator.Record(fields)
 	}
-	dump := aggregator.Dump(int64(30))
+	dump := aggregator.Dump()
 	log.Infof("%v", dump)
 	a := dump["Test_getTest_cost,upstream=127.0.0.1"].(map[string]float64)
 	if a["cnt"] != 10 {
@@ -115,7 +106,4 @@ func TestDump(*testing.T) {
 	if a["p50"] != 4 {
 		panic(a)
 	}
-	if dump["timestamp"].(int64) != 30 {
-		panic(dump)
-	}
 }

peck_task.go

@@ -2,6 +2,7 @@ package logpeck
 
 import (
 	"errors"
+	"time"
 
 	log "github.com/Sirupsen/logrus"
 )
@@ -15,6 +16,7 @@ type PeckTask struct {
 	extractor  Extractor
 	sender     Sender
 	aggregator *Aggregator
+	postTime   int64
 }
 
 // NewPeckTask .
@@ -48,6 +50,9 @@ func NewPeckTask(c *PeckTaskConfig, s *PeckTaskStat) (*PeckTask, error) {
 		sender:     sender,
 		aggregator: aggregator,
 	}
+	if aggregator.IsEnable() {
+		go task.tryDumpAggragator()
+	}
 	log.Infof("[PeckTask] new peck task %#v", task)
 	return task, nil
 }
@@ -87,12 +92,7 @@ func (p *PeckTask) Process(content string) {
 
 	fields, _ := p.extractor.Extract(content)
 	if p.aggregator.IsEnable() {
-		timestamp := p.aggregator.Record(fields)
-		deadline := p.aggregator.IsDeadline(timestamp)
-		if deadline {
-			fields = p.aggregator.Dump(timestamp)
-			p.sender.Send(fields)
-		}
+		p.aggregator.Record(fields)
 	} else {
 		p.sender.Send(fields)
 	}
@@ -109,3 +109,20 @@ func (p *PeckTask) ProcessTest(content string) (map[string]interface{}, error) {
 	}
 	return fields, nil
 }
+
+func (p *PeckTask) tryDumpAggragator() {
+	for {
+		now := time.Now().Unix()
+		interval := p.aggregator.config.Interval
+		if now/interval != p.postTime/interval {
+			fields := p.aggregator.Dump()
+			if len(fields) > 0 {
+				fields["timestamp"] = now
+				p.sender.Send(fields)
+			}
+			p.postTime = now
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+
+}

version.go

@@ -1,4 +1,4 @@
 package logpeck
 
 // VersionString .
-const VersionString string = "0.6.0"
+const VersionString string = "0.7.0"