Compare commits

..

No commits in common. "master" and "nats" have entirely different histories.
master ... nats

16 changed files with 28 additions and 212 deletions

View file

@ -3,9 +3,6 @@ FROM scratch
MAINTAINER Michele Fadda "<mikif70@gmail.com>" MAINTAINER Michele Fadda "<mikif70@gmail.com>"
ARG VER COPY lastlogin_mongodb-v4.3.3 /bin/lastlogin_mongodb
ENV VER ${VER:-0.0.0}
COPY lastlogin_mongodb-${VER} /bin/lastlogin_mongodb
ENTRYPOINT [ "/bin/lastlogin_mongodb" ] ENTRYPOINT [ "/bin/lastlogin_mongodb" ]

View file

@ -1,3 +1,3 @@
#!/bin/bash #!/bin/bash
docker build -t mikif70/llmongo:$(git -C .. describe --tags) -t repo.ism.tiscali.sys:5000/mikif70/llmongo:$(git -C .. describe --tags) --build-arg VER=$(git -C .. describe --tags) . docker build -t mikif70/llmongo:4.3.3 .

View file

@ -1,5 +1,3 @@
#!/bin/bash #!/bin/bash
CGO_ENABLED=0 GOOS=linux go build -o lastlogin_mongodb-$(git describe --tags) -a -installsuffix cgo . CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo .
mv lastlogin_mongodb-$(git describe --tags) Docker/

View file

@ -1,44 +0,0 @@
package main
import (
"fmt"
"log"
"time"
"github.com/hashicorp/consul/api"
)
// Consul client structure
type Consul struct {
RedisTTL time.Duration
Timeout time.Duration
MaxError int
Influxdb string
Month string
Retention float64
MongoURI string
Database string
RedisURI string
Client *api.Client
}
// NewClient New consul client
func NewClient() *Consul {
client, err := api.NewClient(api.DefaultConfig())
if err != nil {
log.Fatalf("Consul error: %+v\n", err)
if opts.Debug {
fmt.Printf("Consul error: %+v\n", err)
}
}
consul := &Consul{
Client: client,
RedisTTL: opts.RedisTTL,
Timeout: opts.Timeout,
MaxError: opts.MaxError,
Influxdb: opts.Influxdb,
}
return consul
}

View file

@ -12,8 +12,7 @@ import (
"strings" "strings"
"time" "time"
"github.com/globalsign/mgo" "gopkg.in/mgo.v2"
// "gopkg.in/mgo.v2"
) )
type consumed struct { type consumed struct {
@ -81,18 +80,6 @@ func consumer() {
continue continue
} }
// se la data e' piu vecchia di RETENTION (15552000 sec) la scarta
if time.Since(time.Unix(date, 0)).Seconds()-opts.Retention >= 0 {
log.Printf("Date Warning: %+v - %s - %s\n", time.Unix(date, 0), prod.user, login)
if opts.Debug {
fmt.Printf("Date Warning: %+v - %s - %s\n", time.Unix(date, 0), prod.user, login)
}
if !opts.Test {
continue
}
}
// verifica se esiste la country // verifica se esiste la country
if len(sval) <= 3 { if len(sval) <= 3 {
sval = append(sval, "NONE") sval = append(sval, "NONE")
@ -130,7 +117,7 @@ func consumer() {
for _, val := range bulk { for _, val := range bulk {
for j, bl := range val { for j, bl := range val {
result, err := bl.Run() _, err := bl.Run()
if j == 0 { if j == 0 {
if err != nil { if err != nil {
if !strings.Contains(err.Error(), "E11000") { if !strings.Contains(err.Error(), "E11000") {
@ -140,7 +127,7 @@ func consumer() {
tipo: "err", tipo: "err",
val: len(prod.logins), val: len(prod.logins),
} }
if opts.Test || opts.Debug { if opts.Test {
log.Printf("ERR: %s - %+v\n", prod.user, prod.logins) log.Printf("ERR: %s - %+v\n", prod.user, prod.logins)
} }
continue continue
@ -149,19 +136,12 @@ func consumer() {
tipo: "dup", tipo: "dup",
val: strings.Count(err.Error(), "E11000"), val: strings.Count(err.Error(), "E11000"),
} }
if opts.Debug {
log.Printf("DUP: %s - %+v\n", prod.user, prod.logins)
}
} }
} else { } else {
if opts.Test { if opts.Test {
log.Printf("OK: %s - %+v\n", prod.user, prod.logins) log.Printf("OK: %s - %+v\n", prod.user, prod.logins)
log.Printf("BulkResult: %s - %+v\n", prod.user, result)
}
counter <- Counterchan{
tipo: "ins",
val: len(allLogins),
} }
} }
} }
} }

View file

@ -6,7 +6,6 @@ import (
"time" "time"
) )
// Counterchan counter structure
type Counterchan struct { type Counterchan struct {
tipo string tipo string
val int val int
@ -39,7 +38,6 @@ func NewCounter() *Counter {
} }
} }
// Run open channel to receive counter
func (c *Counter) Run() { func (c *Counter) Run() {
for { for {
tocount := <-counter tocount := <-counter

View file

@ -1,15 +1,8 @@
// v1.2
var db = connect("192.168.0.1:27017/lastlogin"); var db = connect("192.168.0.1:27017/lastlogin");
//var conn = new Mongo("192.168.0.1");
//var db = conn.getDB("lastlogin");
var dd = function(x) { var s = x.toString(); if(s.length > 1){return s;} else { return "0"+s;}} var dd = function(x) { var s = x.toString(); if(s.length > 1){return s;} else { return "0"+s;}}
var now = new Date(); var now = new Date();
var col = now.getFullYear() + dd(now.getMonth()+1) + dd(now.getDate()) var col = now.getFullYear() + dd(now.getMonth()+1) + dd(now.getDate())
print("creating index: ll_",col.slice(2)); print("creating index: ll_"col.slice(2));
db.getCollection("ll_"+col.slice(2)).createIndexes([{"date": 1},{"country": 1},{"user": 1,"date": -1}]) db.getCollection("ll_"+col.slice(2)).createIndexes([{"date": 1},{"country": 1},{"user": 1,"date": -1}])
db.getCollection("ll_"+col.slice(2)).createIndex({"date": -1},{expireAfterSeconds: 15552000, name: "expire"})
var db = connect("192.168.0.1:27017/katamail");
var dd = function(x) { var s = x.toString(); if(s.length > 1){return s;} else { return "0"+s;}}
var now = new Date();
var col = now.getFullYear() + dd(now.getMonth()+1) + dd(now.getDate())
print("creating index: ll_",col.slice(2));
db.getCollection("ll_"+col.slice(2)).createIndexes([{"date": 1},{"country": 1},{"user": 1,"date": -1}])
db.getCollection("ll_"+col.slice(2)).createIndex({"date": -1},{expireAfterSeconds: 15552000, name: "expire"})

3
dbs.go
View file

@ -10,8 +10,7 @@ import (
"github.com/garyburd/redigo/redis" "github.com/garyburd/redigo/redis"
"github.com/globalsign/mgo" "gopkg.in/mgo.v2"
// "gopkg.in/mgo.v2"
) )
var ( var (

View file

@ -1,18 +0,0 @@
version: '2'
services:
mongodb:
image: mikif70/mongodb:3.6.9
ports:
- 27017:27017
container_name: ll_mongod
volumes:
- ./mongod:/data
depends_on:
- redis
redis:
image: "redis:alpine"
container_name: ll_redis
ports:
- 6379:6379
volumes:
- ./redis:/data

View file

@ -1,9 +0,0 @@
#!/bin/bash
if [ -z $1 ]; then
HOST=192.168.0.1:27017
else
HOST=${1}
fi
docker run --rm -it mikif70/mongotools:3.6.9 mongo ${HOST}

View file

@ -1,16 +0,0 @@
#!/bin/bash
echo "Starting mikif70/llmongo:$(git -C .. describe --tags)"
docker run \
--rm \
-v /opt/WORK/PROJECTS/New_Mail/lastlogin_mongodb/docker-compose/llmongo:/data \
--name llmongo \
mikif70/llmongo:$(git -C .. describe --tags) \
-l /data/llmongo.log \
-r 192.168.0.1:6379 \
-m 192.168.0.1:27017 \
-d lastlogin \
-i test@10.39.253.206:8086 \
-T 80s \
$@

View file

@ -3,7 +3,6 @@ package main
import ( import (
"fmt" "fmt"
"log"
"time" "time"
influxdb "github.com/influxdata/influxdb/client/v2" influxdb "github.com/influxdata/influxdb/client/v2"
@ -16,8 +15,7 @@ var (
func writeStats(start time.Time) { func writeStats(start time.Time) {
if opts.Debug { if opts.Debug {
fmt.Printf("writing to influxdb server: %s\n", opts.Influxdb) fmt.Printf("writing to influxdb server: %s", opts.Influxdb)
fmt.Printf("host: %s -- addr: %s\n", infhost, infdb)
} }
c, err := influxdb.NewHTTPClient(influxdb.HTTPConfig{ c, err := influxdb.NewHTTPClient(influxdb.HTTPConfig{
@ -25,8 +23,7 @@ func writeStats(start time.Time) {
Timeout: 2 * time.Second, Timeout: 2 * time.Second,
}) })
if err != nil { if err != nil {
fmt.Printf("InfluxDB connect Error: %+v\n", err) fmt.Printf("Error: %+v\n", err)
log.Printf("InfluxDB connect Error: %+v\n", err)
return return
} }
defer c.Close() defer c.Close()
@ -36,8 +33,7 @@ func writeStats(start time.Time) {
Precision: "s", Precision: "s",
}) })
if err != nil { if err != nil {
fmt.Printf("InfluxDB batch Error: %+v\n", err) fmt.Printf("Error: %+v\n", err)
log.Printf("InfluxDB batch Error: %+v\n", err)
return return
} }
@ -52,22 +48,12 @@ func writeStats(start time.Time) {
} }
pt, err := influxdb.NewPoint("ll2mongo", tags, fields, start) pt, err := influxdb.NewPoint("ll2mongo", tags, fields, start)
if err != nil { if err != nil {
fmt.Printf("InfluxDB point Error: %+v\n", err) fmt.Printf("Error: %+v\n", err)
log.Printf("InfluxDB point Error: %+v\n", err)
return return
} }
if opts.Debug {
fmt.Printf("InfluxDB pt: %+v\n", pt)
}
bp.AddPoint(pt) bp.AddPoint(pt)
// Write the batch // Write the batch
err = c.Write(bp) c.Write(bp)
if err != nil {
fmt.Printf("InfluxDB write Error: %+v\n", err)
log.Printf("InfluxDB write Error: %+v\n", err)
return
}
} }

10
main.go
View file

@ -14,7 +14,7 @@ import (
) )
const ( const (
_Version = "v5.0.0b6" _Version = "v4.3.3"
_Producer = 0 _Producer = 0
_Consumer = 1 _Consumer = 1
_Remover = 2 _Remover = 2
@ -51,15 +51,11 @@ func main() {
} }
if opts.Influxdb != "" { if opts.Influxdb != "" {
var re = regexp.MustCompile(`(?m)(\w+)@([\w.]+):(\d+)`) re, _ := regexp.Compile(`(\w+)@(\d+.\d+.\d+.\d+:\d+)`)
// re, _ := regexp.Compile(`(\w+)@(\d+.\d+.\d+.\d+:\d+)`)
if re.MatchString(opts.Influxdb) { if re.MatchString(opts.Influxdb) {
match := re.FindStringSubmatch(opts.Influxdb) match := re.FindStringSubmatch(opts.Influxdb)
if opts.Debug {
fmt.Printf("Influxdb match: %+v\n", match)
}
infhost = match[1] infhost = match[1]
infdb = match[2] + ":" + match[3] infdb = match[2]
} else { } else {
opts.Influxdb = "" opts.Influxdb = ""
} }

View file

@ -1,34 +0,0 @@
################
var dd = function(x) { var s = x.toString(); if(s.length > 1){return s;} else { return "0"+s;}}
for (var i=1; i<18; i++) {
var col = "ll_1806"+dd(i);
print(col);
db.getCollection(col).createIndex({date: -1}, {expireAfterSeconds: 15552000, name: "expire", background: true});
}
###########
var dd = function(x) { var s = x.toString(); if(s.length > 1){return s;} else { return "0"+s;}}
for (var m=1; m<=12; m++) {
for (var d=1; d<=31; d++) {
var col = "ll_15"+dd(m)+dd(d);
if (db.getCollection(col).exists != null) {
print(col);
print(db.getCollection(col).totalSize());
}
}
}
####
var ar = db.getCollectionNames()
var len = ar.length
for (var i=0; i<len; i++) { if (ar[i].startsWith("ll")) { print(db.getCollection(ar[i]))}}
####
db.lastlogin_1703.createIndex({date: -1}, {expireAfterSeconds: 15552000, name: "expire", background: true})

View file

@ -1,3 +1,4 @@
// options
package main package main
import ( import (
@ -24,10 +25,8 @@ type Options struct {
MaxError int MaxError int
Influxdb string Influxdb string
Month string Month string
Queue int // Pidfile string
Retention float64 Queue int
Consul string
Port int
} }
var ( var (
@ -46,10 +45,7 @@ func usage() {
-l <logfile> -l <logfile>
-T <running timeout> -T <running timeout>
-i <influxdb [localname@ip:port]> -i <influxdb [localname@ip:port]>
-C <consul [ip:port]>
-P <check port> [port] ## used by consul to check services ##
-q <parallel consumer> -q <parallel consumer>
-R <retention>
-v <version> -v <version>
-D <debug> -D <debug>
-DD <Test>`) -DD <Test>`)
@ -65,6 +61,7 @@ func init() {
} }
opts.LogFile = path.Join(opts.CurrentPath, opts.LogFile) opts.LogFile = path.Join(opts.CurrentPath, opts.LogFile)
// opts.Pidfile = path.Join(opts.CurrentPath, "run", path.Base(os.Args[0])+".pid")
opts.Exe = path.Base(os.Args[0]) opts.Exe = path.Base(os.Args[0])
flag.StringVar(&opts.Influxdb, "i", "", "influxdb server") flag.StringVar(&opts.Influxdb, "i", "", "influxdb server")
@ -72,8 +69,6 @@ func init() {
flag.StringVar(&dbs.Database, "d", "", "Mongodb Database") flag.StringVar(&dbs.Database, "d", "", "Mongodb Database")
flag.StringVar(&dbs.RedisURI, "r", "", "Redis") flag.StringVar(&dbs.RedisURI, "r", "", "Redis")
flag.StringVar(&opts.LogFile, "l", opts.LogFile, "Logs filename") flag.StringVar(&opts.LogFile, "l", opts.LogFile, "Logs filename")
flag.StringVar(&opts.Consul, "C", opts.Consul, "consul client")
flag.IntVar(&opts.Port, "P", 10500, "service check port")
flag.DurationVar(&opts.RedisTTL, "t", opts.RedisTTL, "Redis keys TTL") flag.DurationVar(&opts.RedisTTL, "t", opts.RedisTTL, "Redis keys TTL")
flag.BoolVar(&opts.Version, "v", false, "Version") flag.BoolVar(&opts.Version, "v", false, "Version")
flag.DurationVar(&opts.Timeout, "T", 0, "Running timeout") flag.DurationVar(&opts.Timeout, "T", 0, "Running timeout")
@ -81,5 +76,5 @@ func init() {
flag.BoolVar(&opts.Test, "DD", false, "Test") flag.BoolVar(&opts.Test, "DD", false, "Test")
flag.IntVar(&opts.MaxError, "E", 100, "Max Mongodb Error") flag.IntVar(&opts.MaxError, "E", 100, "Max Mongodb Error")
flag.IntVar(&opts.Queue, "q", 2, "parallel consumer") flag.IntVar(&opts.Queue, "q", 2, "parallel consumer")
flag.Float64Var(&opts.Retention, "R", 15552000, "retention") // flag.StringVar(&opts.Pidfile, "p", opts.Pidfile, "pid file")
} }

View file

@ -24,9 +24,4 @@
# #
# m h dom mon dow command # m h dom mon dow command
### lastlogin import */10 * * * * /opt/llmongo/start.sh > /dev/null
*/3 * * * * /opt/llmongo/run.sh > /dev/null 2>&1
*/5 * * * * /opt/llmongo/katamail.sh > /dev/null 2>&1
05 * * * * /opt/impmongo/run.sh > /dev/null 2>&1
### lastlogin create indexes
00 01 * * * /opt/mongodb/mongo.sh -script /data/day_init.js >> /opt/mongodb/day_init.log 2>&1