141 lines
2.8 KiB
Go
141 lines
2.8 KiB
Go
// consumer
|
|
package main
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"log"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"gopkg.in/mgo.v2"
|
|
)
|
|
|
|
type consumed struct {
|
|
user string
|
|
error bool
|
|
logins []string
|
|
}
|
|
|
|
func hash(val []byte) string {
|
|
|
|
h := sha256.New()
|
|
h.Write(val)
|
|
|
|
return hex.EncodeToString(h.Sum(nil))
|
|
}
|
|
|
|
func consumer() {
|
|
|
|
for {
|
|
|
|
prod := <-consume
|
|
|
|
start := time.Now()
|
|
|
|
status = _Consumer
|
|
|
|
var bulk = make(map[string][]*mgo.Bulk)
|
|
var allLogins = make(map[string]MongoLogin)
|
|
|
|
for i := range prod.logins {
|
|
login := prod.logins[i]
|
|
// se la riga di login e' vuota
|
|
if login == "" {
|
|
log.Println("Login empty: ", prod.user)
|
|
continue
|
|
}
|
|
sval := strings.Split(login, ":")
|
|
// se il formato della riga di login non e' corretto
|
|
if sval[1] == "" {
|
|
log.Println("Login format error: ", login, prod.user)
|
|
continue
|
|
}
|
|
// se il timestamp della riga di login non e' corretto
|
|
date, err := strconv.ParseInt(sval[1], 10, 64)
|
|
if err != nil {
|
|
log.Printf("Date Error: %+v - %s - %s\n", err, prod.user, login)
|
|
continue
|
|
}
|
|
mlID := hash([]byte(fmt.Sprintf("%s%s%s", prod.user, time.Unix(date, 0).Format("20060102T15"), sval[2]))) // Format("20060102T150405")
|
|
ml := MongoLogin{
|
|
// genera l' _ID con user e timestamp
|
|
ID: mlID,
|
|
User: prod.user,
|
|
Protocol: sval[0],
|
|
IP: sval[2],
|
|
Date: time.Unix(date, 0),
|
|
Insert: time.Now(),
|
|
}
|
|
|
|
allLogins[mlID] = ml
|
|
|
|
}
|
|
|
|
for _, val := range allLogins {
|
|
dt := fmt.Sprintf("lastlogin_%s", val.Date.Format("0601"))
|
|
if _, ok := bulk[dt]; !ok {
|
|
for j := range dbs.mdb {
|
|
b := dbs.mdb[j].DB("lastlogin").C(dt).Bulk()
|
|
b.Unordered()
|
|
bulk[dt] = append(bulk[dt], b)
|
|
}
|
|
}
|
|
for _, bl := range bulk[dt] {
|
|
bl.Insert(val)
|
|
}
|
|
}
|
|
|
|
for _, val := range bulk {
|
|
for j, bl := range val {
|
|
_, err := bl.Run()
|
|
if j == 0 {
|
|
if err != nil {
|
|
if !strings.Contains(err.Error(), "E11000") {
|
|
fmt.Printf("Err: %+v\n", err)
|
|
prod.err = true
|
|
counter <- Counterchan{
|
|
tipo: "err",
|
|
val: len(prod.logins),
|
|
}
|
|
if opts.Test {
|
|
log.Printf("ERR: %s - %+v\n", prod.user, prod.logins)
|
|
}
|
|
continue
|
|
} else {
|
|
counter <- Counterchan{
|
|
tipo: "dup",
|
|
val: strings.Count(err.Error(), "E11000"),
|
|
}
|
|
}
|
|
} else {
|
|
if opts.Test {
|
|
log.Printf("OK: %s - %+v\n", prod.user, prod.logins)
|
|
}
|
|
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if opts.Debug {
|
|
fmt.Printf("CONS: user=%s logins=%d in %v - active=%d\n", prod.user, len(prod.logins), time.Since(start), dbs.rdb.ActiveCount())
|
|
}
|
|
|
|
if opts.Debug && len(prod.logins) > 10 {
|
|
fmt.Printf("LOGS: %+v\n", prod.logins)
|
|
}
|
|
|
|
if !prod.err {
|
|
counter <- Counterchan{
|
|
tipo: "rem",
|
|
val: len(prod.logins),
|
|
}
|
|
}
|
|
|
|
// wg.Done()
|
|
remove <- prod
|
|
}
|
|
}
|