aggiunta la possibilita' di scrivere su piu' mongodb

consumer.go

@@ -2,10 +2,9 @@
 package main
 
 import (
-	"fmt"
-	//	"github.com/garyburd/redigo/redis"
 	"crypto/sha256"
 	"encoding/hex"
+	"fmt"
 	"log"
 	"strconv"
 	"strings"
@@ -39,23 +38,13 @@ func consumer() {
 
 		status = _Consumer
 
-		var bulk = make(map[string]*mgo.Bulk)
+		var bulk = make(map[string][]*mgo.Bulk)
-		var rtbulk []MongoLogin
+		var allLogins = make(map[string]MongoLogin)
-		var col = make(map[string]*mgo.Collection)
-		var slogin = make(map[string][]string)
-
-		if opts.Bulk {
-			bulk[opts.Month] = dbs.ll.Bulk()
-			bulk[opts.Month].Unordered()
-		} else {
-			col[opts.Month] = dbs.ll
-		}
 
 		cons := consumed{
 			user:   prod.user,
 			logins: make([]string, 0),
 			error:  false,
-			empty:  true,
 		}
 
 		for i := range prod.logins {
@@ -80,9 +69,10 @@ func consumer() {
 				cons.logins = append(cons.logins, 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:       hash([]byte(fmt.Sprintf("%s%s%s", prod.user, time.Unix(date, 0).Format("20060102T15"), sval[2]))), // Format("20060102T150405")
+				ID:       mlID,
 				User:     prod.user,
 				Protocol: sval[0],
 				IP:       sval[2],
@@ -90,137 +80,38 @@ func consumer() {
 				Insert:   time.Now(),
 			}
 
-			if opts.Month != ml.Date.Format("0601") {
+			allLogins[mlID] = ml
-				dt := fmt.Sprintf("lastlogin_%s", ml.Date.Format("0601"))
-				if opts.Bulk {
-					if dbs.isMongodb() {
-						if _, ok := bulk[dt]; !ok {
-							bulk[dt] = dbs.mdb.DB("lastlogin").C(dt).Bulk()
-							bulk[dt].Unordered()
-						}
-						bulk[dt].Insert(ml)
-						slogin[dt] = append(slogin[dt], login)
-					}
-					if dbs.isRethink() {
-						rtbulk = append(rtbulk, ml)
-						slogin["rt"] = append(slogin["rt"], login)
-					}
-				} else {
-					if dbs.isMongodb() {
-						if _, ok := col[dt]; !ok {
-							col[dt] = dbs.mdb.DB("lastlogin").C(dt)
-						}
-						err = col[dt].Insert(ml)
-						if err != nil {
-							if !strings.Contains(err.Error(), "E11000") {
-								fmt.Printf("Mongo Insert Err: %+v\n", err)
-								cons.error = true
-								counter <- Counterchan{
-									tipo: "err",
-									val:  1,
-								}
-								continue
-							} else {
-								counter <- Counterchan{
-									tipo: "dup",
-									val:  1,
-								}
-							}
-						}
-					}
-					if dbs.isRethink() {
-						_, err = dbs.rtdb.Insert(ml)
-						if err != nil {
-							if !strings.Contains(err.Error(), "Duplicate primary key") {
-								fmt.Printf("RT Insert Err: %+v\n", err)
-								cons.error = true
-								counter <- Counterchan{
-									tipo: "err",
-									val:  1,
-								}
-								continue
-							} else {
-								counter <- Counterchan{
-									tipo: "dup",
-									val:  1,
-								}
-							}
-						}
-					}
 
-					cons.logins = append(cons.logins, login)
+		}
-				}
-			} else {
-				// inserisce il login su Mongodb
-				if opts.Bulk {
-					if dbs.isMongodb() {
-						bulk[opts.Month].Insert(ml)
-						slogin[opts.Month] = append(slogin[opts.Month], login)
-					}
-					if dbs.isRethink() {
-						rtbulk = append(rtbulk, ml)
-						slogin["rt"] = append(slogin["rt"], login)
-					}
-				} else {
-					if dbs.isMongodb() {
-						err = col[opts.Month].Insert(ml)
-						if err != nil {
-							if !strings.Contains(err.Error(), "E11000") {
-								fmt.Printf("Err: %+v\n", err)
-								cons.error = true
-								counter <- Counterchan{
-									tipo: "err",
-									val:  1,
-								}
-								continue
-							} else {
-								counter <- Counterchan{
-									tipo: "dup",
-									val:  1,
-								}
-							}
-						}
-					}
-					if dbs.isRethink() {
-						resp, err := dbs.rtdb.Insert(ml)
-						counter <- Counterchan{
-							tipo: "ins",
-							val:  resp.Inserted,
-						}
-						if err != nil {
-							if !strings.Contains(err.Error(), "Duplicate primary key") {
-								fmt.Printf("RT Insert Err: %+v\n", err)
-								cons.error = true
-								counter <- Counterchan{
-									tipo: "err",
-									val:  1,
-								}
-								continue
-							} else {
-								counter <- Counterchan{
-									tipo: "dup",
-									val:  1,
-								}
-							}
-						}
-					}
 
-					cons.logins = append(cons.logins, login)
+		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)
+			}
 		}
 
-		if opts.Bulk {
+		for _, val := range bulk {
-			if dbs.isMongodb() {
+			for j, bl := range val {
-				for key, _ := range bulk {
+				_, err := bl.Run()
-					_, err := bulk[key].Run()
+				if j == 0 {
 					if err != nil {
 						if !strings.Contains(err.Error(), "E11000") {
 							fmt.Printf("Err: %+v\n", err)
 							cons.error = true
 							counter <- Counterchan{
 								tipo: "err",
-								val:  len(slogin[key]),
+								val:  len(prod.logins),
+							}
+							if opts.Test {
+								log.Printf("ERR: %s - %+v\n", prod.user, prod.logins)
 							}
 							continue
 						} else {
@@ -229,33 +120,16 @@ func consumer() {
 								val:  strings.Count(err.Error(), "E11000"),
 							}
 						}
-					}
-					cons.logins = append(cons.logins, slogin[key]...)
-				}
-			}
-			if dbs.isRethink() {
-				resp, err := dbs.rtdb.MultiInsert(rtbulk)
-				counter <- Counterchan{
-					tipo: "ins",
-					val:  resp.Inserted,
-				}
-				if err != nil {
-					if !strings.Contains(err.Error(), "Duplicate primary key") {
-						cons.error = true
-						counter <- Counterchan{
-							tipo: "err",
-							val:  1,
-						}
-						continue
 					} else {
-						counter <- Counterchan{
+						if opts.Test {
-							tipo: "dup",
+							log.Printf("OK: %s - %+v\n", prod.user, prod.logins)
-							val:  1,
 						}
+
 					}
 				}
-				cons.logins = append(cons.logins, slogin["rt"]...)
 			}
+
+			cons.logins = append(cons.logins, prod.logins...)
 		}
 
 		counter <- Counterchan{
@@ -263,10 +137,6 @@ func consumer() {
 			val:  len(prod.logins),
 		}
 
-		if opts.MaxLogins > -1 && len(prod.logins) < opts.MaxLogins {
-			cons.empty = false
-		}
-
 		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())
 		}

dbs.go

@@ -2,15 +2,14 @@
 package main
 
 import (
-	"github.com/garyburd/redigo/redis"
+	//	"fmt"
-	//	"github.com/fzzy/radix/redis"
-	"fmt"
 	"log"
 	"os"
 	"strings"
 	"time"
 
-	// rt "gopkg.in/dancannon/gorethink.v2"
+	"github.com/garyburd/redigo/redis"
+
 	"gopkg.in/mgo.v2"
 )
 
@@ -23,15 +22,11 @@ var (
 
 // Dbs structure
 type Dbs struct {
-	MongoURI   string
+	MongoURI string
-	Database   string
+	Database string
-	RedisURI   string
+	RedisURI string
-	RethinkURI string
+	rdb      *redis.Pool //*redis.Client
-	rdb        *redis.Pool //*redis.Client
+	mdb      []*mgo.Session
-	rtdb       *Rethink
-	mdb        *mgo.Session
-	ll         *mgo.Collection
-	//	us       *mgo.Collection
 }
 
 // MongoLogin structure
@@ -62,14 +57,6 @@ type Index struct {
 	Date time.Time `json:"date"`
 }
 
-func (db *Dbs) isRethink() bool {
-	if db.RethinkURI != "" {
-		return true
-	}
-
-	return false
-}
-
 func (db *Dbs) isMongodb() bool {
 	if db.MongoURI != "" {
 		return true
@@ -78,31 +65,6 @@ func (db *Dbs) isMongodb() bool {
 	return false
 }
 
-func (db *Dbs) poolRethink() {
-	var err error
-
-	if opts.Debug {
-		fmt.Printf("DBS: %+v\n", db)
-	}
-
-	if db.RethinkURI != "" {
-		uri := strings.Split(dbs.RethinkURI, ",")
-		if opts.Debug {
-			fmt.Printf("RT_URI: %s\n", uri)
-		}
-		db.rtdb, err = NewRethinkDB(uri)
-		if err != nil {
-			fmt.Println("RethinkDB connect Error: ", err.Error())
-			os.Exit(-4)
-		}
-	}
-
-	if opts.Debug {
-		fmt.Printf("DBS: %+v\n", db)
-	}
-
-}
-
 func (db *Dbs) poolRedis() {
 
 	dbs.rdb = &redis.Pool{
@@ -126,12 +88,17 @@ func (db *Dbs) poolRedis() {
 }
 
 func (db *Dbs) connectMongo() {
-	var err error
-	db.mdb, err = mgo.Dial(db.MongoURI)
-	if err != nil {
-		log.Println("Mongodb connect Error: ", err.Error())
-		os.Exit(-3)
-	}
 
-	db.ll = db.mdb.DB(db.Database).C(fmt.Sprintf("lastlogin_%s", opts.Month))
+	mongoList := strings.Split(db.MongoURI, ",")
+
+	for m := range mongoList {
+		nm, err := mgo.Dial(mongoList[m])
+		if err != nil {
+			log.Println("Mongodb connect Error: ", err.Error())
+			os.Exit(-3)
+		}
+		nm.SetSocketTimeout(5 * time.Second)
+		nm.SetSyncTimeout(5 * time.Second)
+		db.mdb = append(db.mdb, nm)
+	}
 }

main.go

@@ -11,7 +11,7 @@ import (
 )
 
 const (
-	_Version  = "v3.2.0"
+	_Version  = "v4.0.0"
 	_Producer = 0
 	_Consumer = 1
 	_Remover  = 2
@@ -74,14 +74,9 @@ func main() {
 	dbs.poolRedis()
 	defer dbs.rdb.Close()
 
-	if dbs.isMongodb() {
+	dbs.connectMongo()
-		dbs.connectMongo()
+	for k := range dbs.mdb {
-		defer dbs.mdb.Close()
+		defer dbs.mdb[k].Close()
-	}
-
-	if dbs.isRethink() {
-		dbs.poolRethink()
-		defer dbs.rtdb.Close()
 	}
 
 	if opts.Timeout > 0 {
@@ -99,7 +94,9 @@ func main() {
 	go count.Run()
 	go producer()
 	for i := 0; i < opts.Queue; i++ {
-		go consumer()
+		for j := 0; j < len(dbs.mdb); j++ {
+			go consumer()
+		}
 		go remover()
 	}
 

options.go

@@ -2,10 +2,8 @@
 package main
 
 import (
-	//	"encoding/json"
 	"flag"
 	"fmt"
-	//	"io/ioutil"
 	"log"
 	"os"
 	"path"
@@ -21,10 +19,9 @@ type Options struct {
 	LogFile     string
 	ConfigFile  string
 	Timeout     time.Duration
-	MaxLogins   int
 	Debug       bool
+	Test        bool
 	Version     bool
-	Bulk        bool
 	MaxError    int
 	Influxdb    string
 	Hostname    string
@@ -35,9 +32,8 @@ type Options struct {
 
 var (
 	opts = Options{
-		RedisTTL:  time.Hour * 11688, // 16 mesi
+		RedisTTL: time.Hour * 11688, // 16 mesi
-		LogFile:   "log/llmongo.log",
+		LogFile:  "log/llmongo.log",
-		MaxLogins: -1,
 	}
 )
 
@@ -63,14 +59,12 @@ func init() {
 	flag.StringVar(&dbs.MongoURI, "m", "", "Mongodb")
 	flag.StringVar(&dbs.Database, "d", dbs.Database, "Mongodb Database")
 	flag.StringVar(&dbs.RedisURI, "r", dbs.RedisURI, "Redis")
-	flag.StringVar(&dbs.RethinkURI, "R", "", "Rethink DB")
 	flag.StringVar(&opts.LogFile, "l", opts.LogFile, "Logs filename")
-	flag.IntVar(&opts.MaxLogins, "L", opts.MaxLogins, "Max lastlogins")
 	flag.DurationVar(&opts.RedisTTL, "t", opts.RedisTTL, "Redis keys TTL")
 	flag.BoolVar(&opts.Version, "v", false, "Version")
 	flag.DurationVar(&opts.Timeout, "T", 0, "Running timeout")
 	flag.BoolVar(&opts.Debug, "D", false, "Debug")
-	flag.BoolVar(&opts.Bulk, "B", false, "Bulk")
+	flag.BoolVar(&opts.Test, "DD", false, "Test")
 	flag.IntVar(&opts.MaxError, "E", 100, "Max Mongodb Error")
 	flag.IntVar(&opts.Queue, "q", 2, "parallel consumer")
 	flag.StringVar(&opts.Pidfile, "p", opts.Pidfile, "pid file")

producer.go

@@ -4,7 +4,7 @@ package main
 import (
 	"fmt"
 	"log"
-	"strconv"
+	//	"strconv"
 	"time"
 
 	"github.com/garyburd/redigo/redis"
@@ -27,58 +27,52 @@ func producer() {
 		start := time.Now()
 		// estrae un userid dalla lista degli utenti che hanno fatto login
 		user, err := redis.String(conn.Do("spop", "llindex"))
-		//		if opts.Debug {
-		//			log.Printf("SPOP: %+v - %+v\n", user, err)
-		//			fmt.Printf("SPOP: %+v - %+v\n", user, err)
-		//		}
 		// se non ci sono piu' userid esce
 		if err != nil {
 			if opts.Debug {
 				fmt.Printf("LLINDEX empty: %v\n", err)
 			}
 			log.Printf("LLINDEX empty: %v\n", err)
-			//loop[id] = false
-			//done[id] <- true
 			break
 		}
 
-		// estrae <MaxLogins> login dell'utente "user"
+		// estrae tutti i logins dell'utente "user"
-		logs, err := redis.Strings(conn.Do("lrange", user, "1", strconv.Itoa(opts.MaxLogins)))
+		logs, err := redis.Strings(conn.Do("lrange", user, "1", "-1"))
 		if err != nil {
 			if opts.Debug {
 				fmt.Printf("LRANGE: %+v - %+v\n", err, logs)
 			}
 			log.Printf("LRANGE: %+v - %+v\n", err, logs)
 		}
-		//		if opts.Debug {
-		//			fmt.Printf("LRANGE: %s - %d\n", user, len(logs))
-		//			log.Printf("LRANGE: %s - %d\n", user, len(logs))
-		//		}
 
 		if opts.Debug {
 			fmt.Printf("PROD: user=%s login=%d in %v - conn=%d\n", user, len(logs), time.Since(start), dbs.rdb.ActiveCount())
-			//			log.Printf("PROD (%d): user=%s login=%d in %v - conn=%d\n", id, user, len(logs), time.Since(start), dbs.rdb.ActiveCount())
 		}
 
-		counter <- Counterchan{
+		if len(logs) > 0 {
-			tipo: "user",
+			counter <- Counterchan{
-			val:  1,
+				tipo: "user",
-		}
+				val:  1,
-		wg.Add(1)
+			}
-		counter <- Counterchan{
+			wg.Add(1)
-			tipo: "wg",
+			counter <- Counterchan{
-			val:  1,
+				tipo: "wg",
-		}
+				val:  1,
+			}
 
-		if opts.Debug {
+			if opts.Debug {
-			fmt.Printf("PROD: %+v\n", time.Since(start))
+				fmt.Printf("PROD: %+v\n", time.Since(start))
-		}
+			}
 
-		consume <- produced{
+			if opts.Test {
-			user:   user,
+				log.Printf("PROD: %s - %d\n", user, len(logs))
-			logins: logs,
+			}
-		}
 
+			consume <- produced{
+				user:   user,
+				logins: logs,
+			}
+		}
 	}
 
 	done <- true

remover.go

@@ -3,7 +3,7 @@ package main
 
 import (
 	"fmt"
-	//	"log"
+
 	"time"
 )
 
@@ -17,16 +17,16 @@ func remover() {
 
 		status = _Remover
 
-		//		wg.Add(1)
-
 		start := time.Now()
-		for i := range rem.logins {
+		if !rem.error {
-			// cancella da Redis la riga di login inserita partendo da 1
+			for i := range rem.logins {
-			conn.Send("lrem", rem.user, "1", rem.logins[i])
+				// cancella da Redis la riga di login inserita partendo da 1
+				conn.Send("lrem", rem.user, "1", rem.logins[i])
+			}
 		}
 
 		// se ci sono errori o non e' vuota la lista di logins reinserisce lo user
-		if !rem.empty || rem.error {
+		if rem.error {
 			if opts.Debug {
 				fmt.Printf("SADD: %s\n", rem.user)
 			}
@@ -37,9 +37,12 @@ func remover() {
 		}
 		conn.Send("expire", rem.user, opts.RedisTTL.Seconds())
 		conn.Flush()
-		counter <- Counterchan{
+
-			tipo: "rem",
+		if !rem.error {
-			val:  len(rem.logins),
+			counter <- Counterchan{
+				tipo: "rem",
+				val:  len(rem.logins),
+			}
 		}
 
 		if opts.Debug {