MongoDB/llconsolidate
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

aggiunti parametri per il monitoraggio

Browse source
This commit is contained in:
Miki 2016-04-06 12:05:40 +02:00
parent 1ebc85d1c3
commit 34e86dd7fd
3 changed files with 46 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

22
aggregate.go
View file

@ -3,9 +3,10 @@ package main
import ( import (
"fmt" "fmt"
"gopkg.in/mgo.v2/bson"
"log" "log"
"time" "time"
"gopkg.in/mgo.v2/bson"
) )
func aggregate(ys time.Time, ye time.Time) { func aggregate(ys time.Time, ye time.Time) {
@ -14,6 +15,7 @@ func aggregate(ys time.Time, ye time.Time) {
p := dbs.ll.Pipe([]bson.M{{"$match": bson.M{"date": bson.M{"$gte": ys, "$lte": ye}}}, {"$group": bson.M{"_id": "$user"}}, {"$project": bson.M{"user": "$_id"}}}) p := dbs.ll.Pipe([]bson.M{{"$match": bson.M{"date": bson.M{"$gte": ys, "$lte": ye}}}, {"$group": bson.M{"_id": "$user"}}, {"$project": bson.M{"user": "$_id"}}})
idb.Pipe = time.Since(qStart)
if opts.Debug { if opts.Debug {
fmt.Printf("Aggregate user: %s\n\r", time.Since(qStart)) fmt.Printf("Aggregate user: %s\n\r", time.Since(qStart))
} }
@ -25,7 +27,7 @@ func aggregate(ys time.Time, ye time.Time) {
it := p.Iter() it := p.Iter()
for it.Next(&ar) { for it.Next(&ar) {
countTOT += 1 idb.CountTOT += 1
ll := LastLoginDay{} ll := LastLoginDay{}
ll.User = ar.User ll.User = ar.User
ll.Date = ys ll.Date = ys
@ -39,6 +41,13 @@ func aggregate(ys time.Time, ye time.Time) {
nq := dbs.ll.Find(bson.M{"date": bson.M{"$gte": ys, "$lte": ye}, "user": ar.User}).Sort("date") nq := dbs.ll.Find(bson.M{"date": bson.M{"$gte": ys, "$lte": ye}, "user": ar.User}).Sort("date")
//fmt.Printf("User: %s -> %s\n\r", ar[u], time.Since(qStart) ) //fmt.Printf("User: %s -> %s\n\r", ar[u], time.Since(qStart) )
iter := nq.Iter() iter := nq.Iter()
if idb.Find == 0 {
idb.Find = time.Since(qStart)
} else {
idb.Find = (idb.Find + time.Since(qStart)) / 2
}
result := LastLogin{} result := LastLogin{}
ips := []IPs{} ips := []IPs{}
lastip := IPs{} lastip := IPs{}
@ -64,11 +73,18 @@ func aggregate(ys time.Time, ye time.Time) {
} }
ll.IPs = ips ll.IPs = ips
//fmt.Printf("Upsert %+v\n\r", ll) //fmt.Printf("Upsert %+v\n\r", ll)
iStart := time.Now()
_, err := dbs.lc.Upsert(Index{User: ll.User, Date: ll.Date}, ll) _, err := dbs.lc.Upsert(Index{User: ll.User, Date: ll.Date}, ll)
if err != nil { if err != nil {
log.Println("Insert error: ", err) log.Println("Insert error: ", err)
} }
countOK += 1 if idb.Insert == 0 {
idb.Insert = time.Since(iStart)
} else {
idb.Insert = (idb.Insert + time.Since(iStart)) / 2
}
idb.CountOK += 1
} }
} }

20
influxdb.go
View file

@ -8,7 +8,16 @@ import (
influxdb "github.com/influxdata/influxdb/client/v2" influxdb "github.com/influxdata/influxdb/client/v2"
) )
func writeStats(start time.Time, stop time.Duration) { type InfluxdbOutput struct {
CountOK int
CountTOT int
Stop time.Duration
Pipe time.Duration
Find time.Duration
Insert time.Duration
}
func writeStats(start time.Time) {
if opts.Debug { if opts.Debug {
fmt.Printf("writing to influxdb server: %s", opts.Influxdb) fmt.Printf("writing to influxdb server: %s", opts.Influxdb)
} }
@ -34,9 +43,12 @@ func writeStats(start time.Time, stop time.Duration) {
tags := map[string]string{"server": opts.Hostname} tags := map[string]string{"server": opts.Hostname}
fields := map[string]interface{}{ fields := map[string]interface{}{
"LoginOK": countOK, "LoginOK": idb.CountOK,
"LoginTOT": countTOT, "LoginTOT": idb.CountTOT,
"stop": stop.Seconds(), "stop": idb.Stop.Seconds(),
"pipe": idb.Pipe.Seconds(),
"find": idb.Find.Seconds(),
"insert": idb.Insert.Seconds(),
} }
pt, err := influxdb.NewPoint("llday", tags, fields, start) pt, err := influxdb.NewPoint("llday", tags, fields, start)
if err != nil { if err != nil {

18
lastlogin_consolidate.go
View file

@ -12,7 +12,7 @@ import (
) )
const ( const (
_VERSION = "v1.2.1" _VERSION = "v1.2.2"
_tformat = "2006-01-02" _tformat = "2006-01-02"
_24h = (time.Hour * 23) + (time.Minute * 59) + (time.Second * 59) _24h = (time.Hour * 23) + (time.Minute * 59) + (time.Second * 59)
_10m = (time.Minute * 10) _10m = (time.Minute * 10)
@ -30,8 +30,12 @@ var (
dbs = Dbs{} dbs = Dbs{}
countTOT = 0 idb = InfluxdbOutput{
countOK = 0 CountTOT: 0,
CountOK: 0,
Insert: 0,
Find: 0,
}
) )
func main() { func main() {
@ -108,12 +112,12 @@ func main() {
log.Printf("Stop %s: %s\n", ys[i], time.Since(pStart)) log.Printf("Stop %s: %s\n", ys[i], time.Since(pStart))
} }
stop := time.Since(start) idb.Stop = time.Since(start)
fmt.Printf("Stop: OK: %d - TOT: %d - Time: %s\n\r", countOK, countTOT, stop) fmt.Printf("Stop: OK: %d - TOT: %d - Time: %s\n\r", idb.CountOK, idb.CountTOT, idb.Stop)
log.Printf("Stop: OK: %d - TOT: %d - Time: %s\n\r", countOK, countTOT, stop) log.Printf("Stop: OK: %d - TOT: %d - Time: %s\n\r", idb.CountOK, idb.CountTOT, idb.Stop)
if opts.Influxdb != "" { if opts.Influxdb != "" {
writeStats(start, stop) writeStats(start)
} }
} }