llconsolidate/aggregate.go

// 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)
}