Compare commits
No commits in common. "master" and "nats" have entirely different histories.
16 changed files with 28 additions and 212 deletions
|
@ -3,9 +3,6 @@ FROM scratch
|
|||
|
||||
MAINTAINER Michele Fadda "<mikif70@gmail.com>"
|
||||
|
||||
ARG VER
|
||||
ENV VER ${VER:-0.0.0}
|
||||
|
||||
COPY lastlogin_mongodb-${VER} /bin/lastlogin_mongodb
|
||||
COPY lastlogin_mongodb-v4.3.3 /bin/lastlogin_mongodb
|
||||
|
||||
ENTRYPOINT [ "/bin/lastlogin_mongodb" ]
|
||||
|
|
|
@ -1,3 +1,3 @@
|
|||
#!/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 .
|
||||
|
|
4
build.sh
4
build.sh
|
@ -1,5 +1,3 @@
|
|||
#!/bin/bash
|
||||
|
||||
CGO_ENABLED=0 GOOS=linux go build -o lastlogin_mongodb-$(git describe --tags) -a -installsuffix cgo .
|
||||
|
||||
mv lastlogin_mongodb-$(git describe --tags) Docker/
|
||||
CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo .
|
||||
|
|
44
consul.go
44
consul.go
|
@ -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
|
||||
}
|
28
consumer.go
28
consumer.go
|
@ -12,8 +12,7 @@ import (
|
|||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/globalsign/mgo"
|
||||
// "gopkg.in/mgo.v2"
|
||||
"gopkg.in/mgo.v2"
|
||||
)
|
||||
|
||||
type consumed struct {
|
||||
|
@ -81,18 +80,6 @@ func consumer() {
|
|||
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
|
||||
if len(sval) <= 3 {
|
||||
sval = append(sval, "NONE")
|
||||
|
@ -130,7 +117,7 @@ func consumer() {
|
|||
|
||||
for _, val := range bulk {
|
||||
for j, bl := range val {
|
||||
result, err := bl.Run()
|
||||
_, err := bl.Run()
|
||||
if j == 0 {
|
||||
if err != nil {
|
||||
if !strings.Contains(err.Error(), "E11000") {
|
||||
|
@ -140,7 +127,7 @@ func consumer() {
|
|||
tipo: "err",
|
||||
val: len(prod.logins),
|
||||
}
|
||||
if opts.Test || opts.Debug {
|
||||
if opts.Test {
|
||||
log.Printf("ERR: %s - %+v\n", prod.user, prod.logins)
|
||||
}
|
||||
continue
|
||||
|
@ -149,19 +136,12 @@ func consumer() {
|
|||
tipo: "dup",
|
||||
val: strings.Count(err.Error(), "E11000"),
|
||||
}
|
||||
if opts.Debug {
|
||||
log.Printf("DUP: %s - %+v\n", prod.user, prod.logins)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if opts.Test {
|
||||
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),
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -6,7 +6,6 @@ import (
|
|||
"time"
|
||||
)
|
||||
|
||||
// Counterchan counter structure
|
||||
type Counterchan struct {
|
||||
tipo string
|
||||
val int
|
||||
|
@ -39,7 +38,6 @@ func NewCounter() *Counter {
|
|||
}
|
||||
}
|
||||
|
||||
// Run open channel to receive counter
|
||||
func (c *Counter) Run() {
|
||||
for {
|
||||
tocount := <-counter
|
||||
|
|
13
day_init.js
13
day_init.js
|
@ -1,15 +1,8 @@
|
|||
// v1.2
|
||||
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 now = new Date();
|
||||
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)).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
3
dbs.go
|
@ -10,8 +10,7 @@ import (
|
|||
|
||||
"github.com/garyburd/redigo/redis"
|
||||
|
||||
"github.com/globalsign/mgo"
|
||||
// "gopkg.in/mgo.v2"
|
||||
"gopkg.in/mgo.v2"
|
||||
)
|
||||
|
||||
var (
|
||||
|
|
|
@ -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
|
|
@ -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}
|
|
@ -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 \
|
||||
$@
|
24
influxdb.go
24
influxdb.go
|
@ -3,7 +3,6 @@ package main
|
|||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
influxdb "github.com/influxdata/influxdb/client/v2"
|
||||
|
@ -16,8 +15,7 @@ var (
|
|||
|
||||
func writeStats(start time.Time) {
|
||||
if opts.Debug {
|
||||
fmt.Printf("writing to influxdb server: %s\n", opts.Influxdb)
|
||||
fmt.Printf("host: %s -- addr: %s\n", infhost, infdb)
|
||||
fmt.Printf("writing to influxdb server: %s", opts.Influxdb)
|
||||
}
|
||||
|
||||
c, err := influxdb.NewHTTPClient(influxdb.HTTPConfig{
|
||||
|
@ -25,8 +23,7 @@ func writeStats(start time.Time) {
|
|||
Timeout: 2 * time.Second,
|
||||
})
|
||||
if err != nil {
|
||||
fmt.Printf("InfluxDB connect Error: %+v\n", err)
|
||||
log.Printf("InfluxDB connect Error: %+v\n", err)
|
||||
fmt.Printf("Error: %+v\n", err)
|
||||
return
|
||||
}
|
||||
defer c.Close()
|
||||
|
@ -36,8 +33,7 @@ func writeStats(start time.Time) {
|
|||
Precision: "s",
|
||||
})
|
||||
if err != nil {
|
||||
fmt.Printf("InfluxDB batch Error: %+v\n", err)
|
||||
log.Printf("InfluxDB batch Error: %+v\n", err)
|
||||
fmt.Printf("Error: %+v\n", err)
|
||||
return
|
||||
}
|
||||
|
||||
|
@ -52,22 +48,12 @@ func writeStats(start time.Time) {
|
|||
}
|
||||
pt, err := influxdb.NewPoint("ll2mongo", tags, fields, start)
|
||||
if err != nil {
|
||||
fmt.Printf("InfluxDB point Error: %+v\n", err)
|
||||
log.Printf("InfluxDB point Error: %+v\n", err)
|
||||
fmt.Printf("Error: %+v\n", err)
|
||||
return
|
||||
}
|
||||
|
||||
if opts.Debug {
|
||||
fmt.Printf("InfluxDB pt: %+v\n", pt)
|
||||
}
|
||||
|
||||
bp.AddPoint(pt)
|
||||
|
||||
// Write the batch
|
||||
err = c.Write(bp)
|
||||
if err != nil {
|
||||
fmt.Printf("InfluxDB write Error: %+v\n", err)
|
||||
log.Printf("InfluxDB write Error: %+v\n", err)
|
||||
return
|
||||
}
|
||||
c.Write(bp)
|
||||
}
|
||||
|
|
10
main.go
10
main.go
|
@ -14,7 +14,7 @@ import (
|
|||
)
|
||||
|
||||
const (
|
||||
_Version = "v5.0.0b6"
|
||||
_Version = "v4.3.3"
|
||||
_Producer = 0
|
||||
_Consumer = 1
|
||||
_Remover = 2
|
||||
|
@ -51,15 +51,11 @@ func main() {
|
|||
}
|
||||
|
||||
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) {
|
||||
match := re.FindStringSubmatch(opts.Influxdb)
|
||||
if opts.Debug {
|
||||
fmt.Printf("Influxdb match: %+v\n", match)
|
||||
}
|
||||
infhost = match[1]
|
||||
infdb = match[2] + ":" + match[3]
|
||||
infdb = match[2]
|
||||
} else {
|
||||
opts.Influxdb = ""
|
||||
}
|
||||
|
|
|
@ -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})
|
||||
|
15
options.go
15
options.go
|
@ -1,3 +1,4 @@
|
|||
// options
|
||||
package main
|
||||
|
||||
import (
|
||||
|
@ -24,10 +25,8 @@ type Options struct {
|
|||
MaxError int
|
||||
Influxdb string
|
||||
Month string
|
||||
Queue int
|
||||
Retention float64
|
||||
Consul string
|
||||
Port int
|
||||
// Pidfile string
|
||||
Queue int
|
||||
}
|
||||
|
||||
var (
|
||||
|
@ -46,10 +45,7 @@ func usage() {
|
|||
-l <logfile>
|
||||
-T <running timeout>
|
||||
-i <influxdb [localname@ip:port]>
|
||||
-C <consul [ip:port]>
|
||||
-P <check port> [port] ## used by consul to check services ##
|
||||
-q <parallel consumer>
|
||||
-R <retention>
|
||||
-v <version>
|
||||
-D <debug>
|
||||
-DD <Test>`)
|
||||
|
@ -65,6 +61,7 @@ func init() {
|
|||
}
|
||||
|
||||
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])
|
||||
|
||||
flag.StringVar(&opts.Influxdb, "i", "", "influxdb server")
|
||||
|
@ -72,8 +69,6 @@ func init() {
|
|||
flag.StringVar(&dbs.Database, "d", "", "Mongodb Database")
|
||||
flag.StringVar(&dbs.RedisURI, "r", "", "Redis")
|
||||
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.BoolVar(&opts.Version, "v", false, "Version")
|
||||
flag.DurationVar(&opts.Timeout, "T", 0, "Running timeout")
|
||||
|
@ -81,5 +76,5 @@ func init() {
|
|||
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.Float64Var(&opts.Retention, "R", 15552000, "retention")
|
||||
// flag.StringVar(&opts.Pidfile, "p", opts.Pidfile, "pid file")
|
||||
}
|
||||
|
|
|
@ -24,9 +24,4 @@
|
|||
#
|
||||
# m h dom mon dow command
|
||||
|
||||
### lastlogin import
|
||||
*/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
|
||||
*/10 * * * * /opt/llmongo/start.sh > /dev/null
|
||||
|
|
Loading…
Add table
Reference in a new issue