// aggregate package main import ( "fmt" "log" // "sort" "time" "gopkg.in/mgo.v2/bson" ) func bulkWrite() { _, err := dbs.bulk.Run() if err != nil { log.Println("Insert error: ", err) } //fmt.Printf("Bulk res: %+v\n", res) } func consolidate(user Users, ys time.Time, ye time.Time) { if opts.Bulk { dbs.bulk = dbs.lc.Bulk() dbs.bulk.Unordered() } idb.CountTOT += 1 ll := LastLoginDay{} ll.User = user.User ll.Date = ys // DEBUG logins := user.Logins ips := []IPs{} lastip := IPs{} for l := range logins { if logins[l].IP == lastip.IP && logins[l].Date.Sub(lastip.Date) < opts.Interval { /* if opts.Debug { fmt.Println("IPs: ", logins[l].IP, logins[l].Date, logins[l].Date.Sub(lastip.Date)) } */ } else { ips = append(ips, IPs{IP: logins[l].IP, Date: logins[l].Date, Protocol: logins[l].Protocol}) lastip.Date = logins[l].Date lastip.IP = logins[l].IP } switch logins[l].Protocol { case "pop3", "pop": ll.Protocols.Pop += 1 case "imap": ll.Protocols.Imap += 1 case "web": ll.Protocols.Web += 1 } } ll.IPs = ips iStart := time.Now() if opts.Bulk { //dbs.bulk.Upsert(Index{User: ll.User, Date: ll.Date}, ll) dbs.bulk.Insert(ll) } else { _, err := dbs.lc.Upsert(Index{User: ll.User, Date: ll.Date}, ll) if err != nil { log.Println("Insert error: ", err) } // fmt.Printf("Change: %+v\n", info) } idb.Insert += time.Since(iStart) idb.CountOK += 1 if opts.Bulk { bulkWrite() } // if opts.Debug { // fmt.Printf("Insert: %d in %v\n", idb.CountOK, idb.Insert) // } } func aggregate(ys time.Time, ye time.Time) { groups := []string{"[^a-z]", "[be]", "[rv]", "[dt]", "[li]", "[pzjkwxy]", "[fn]", "[co]", "[gu]", "[sh]", "[aq]", "[m]"} for g := range groups { qStart := time.Now() p := dbs.ll.Pipe([]bson.M{ {"$match": bson.M{"date": bson.M{"$gte": ys, "$lte": ye}, "user": bson.RegEx{"^" + groups[g], ""}}}, {"$sort": bson.M{"user": -1, "date": 1}}, {"$group": bson.M{"_id": "$user", "logins": bson.M{"$push": bson.M{"protocol": "$protocol", "date": "$date", "ip": "$ip"}}}}}).AllowDiskUse() iter := p.Iter() defer iter.Close() var result Users for iter.Next(&result) { consolidate(result, ys, ye) } if opts.Debug { fmt.Printf("Group %v: %+v\n", groups[g], time.Since(qStart)) } // p.All(&result) idb.Pipe = idb.Pipe + time.Since(qStart) } fmt.Printf("Date: %s - %s\n", ys, ye) log.Printf("Date: %s - %s\n", ys, ye) }