updated ipbl

* removed unused columns in tables
* added multithreaded support for ip scans
* added health endpoint

src/models/as.go

@@ -1,6 +1,10 @@
 package models
 
-import "xorm.io/xorm"
+import (
+	"time"
+
+	"xorm.io/xorm"
+)
 
 func (as *AutonomousSystem) GetOrCreate(session *xorm.Session) (apias *APIAutonomousSystem, err error) {
 	has, err := session.Get(as)
@@ -47,6 +51,8 @@ type AutonomousSystem struct {
 	ID      int       `xorm:"pk autoincr"`
 	ASID    int       `xorm:"integer unique(asindex) as_id"`
 	ASName  string    `xorm:"text unique(asindex) as_name"`
+	Created time.Time `xorm:"created notnull"`
+	Updated time.Time `xorm:"updated notnull"`
 }
 
 type APIAutonomousSystem struct {

src/models/city.go

@@ -1,6 +1,8 @@
 package models
 
 import (
+	"time"
+
 	"xorm.io/xorm"
 )
 
@@ -46,6 +48,8 @@ func (city *City) APIParse(apicity APICity) (err error) {
 type City struct {
 	ID       int       `xorm:"pk autoincr"`
 	CityName string    `xorm:"text unique(cityindex) city_name" json:"city_name"`
+	Created  time.Time `xorm:"created notnull"`
+	Updated  time.Time `xorm:"updated notnull"`
 }
 
 type APICity struct {

src/models/country.go

@@ -1,6 +1,10 @@
 package models
 
-import "xorm.io/xorm"
+import (
+	"time"
+
+	"xorm.io/xorm"
+)
 
 func (country *Country) GetOrCreate(session *xorm.Session) (apicountry *APICountry, err error) {
 	has, err := session.Get(country)
@@ -44,6 +48,8 @@ func (country *Country) APIParse(apicountry APICountry) (err error) {
 type Country struct {
 	ID          int       `xorm:"pk autoincr"`
 	CountryName string    `xorm:"text unique(countryindex) country_name" json:"country_name"`
+	Created     time.Time `xorm:"created notnull"`
+	Updated     time.Time `xorm:"updated notnull"`
 }
 
 type APICountry struct {

src/models/event.go

@@ -1,7 +1,6 @@
 package models
 
 import (
-	"database/sql"
 	"time"
 
 	"git.paulbsd.com/paulbsd/ipbl/src/config"
@@ -34,19 +33,19 @@ func (event *Event) APIFormat() *APIEvent {
 func (event *Event) APIParse(session *xorm.Session, apievent APIEvent) (err error) {
 	*event = Event{
 		IP:   &IP{IP: apievent.IPData.IP},
-		Src: apievent.IPData.Src,
-		Hostname: sql.NullString{
-			String: apievent.IPData.Hostname,
-			Valid:  true},
+		Host: &Host{Host: apievent.IPData.Hostname},
+		Src:  &Src{Src: apievent.IPData.Src},
 	}
 	event.IP.GetOrCreate(session)
+	event.Host.GetOrCreate(session)
+	event.Src.GetOrCreate(session)
 	return
 }
 
 type Event struct {
 	ID      int       `xorm:"pk autoincr"`
-	Src      string         `xorm:"text notnull"`
-	Hostname sql.NullString `xorm:"text default NULL"`
+	Src     *Src      `xorm:"int src_id"`
+	Host    *Host     `xorm:"int host_id"`
 	IP      *IP       `xorm:"int ip_id"`
 	Created time.Time `xorm:"created notnull"`
 }

src/models/host.go

@@ -1,6 +1,8 @@
 package models
 
 import (
+	"time"
+
 	"xorm.io/xorm"
 )
 
@@ -46,6 +48,8 @@ func (host *Host) APIParse(apihost APIHost) (err error) {
 type Host struct {
 	ID      int       `xorm:"pk autoincr"`
 	Host    string    `xorm:"text unique host" json:"host"`
+	Created time.Time `xorm:"created notnull"`
+	Updated time.Time `xorm:"updated notnull"`
 }
 
 type APIHost struct {

src/models/hostinfo.go

@@ -36,7 +36,6 @@ func ProcessHostInfo(cfg *config.Config, hostinfos []HostInfo) (err error) {
 func HostInfoToIP(session *xorm.Session, hi *HostInfo) (ip IP) {
 	ip = IP{
 		IP: hi.IP,
-		Whois: hi.Whois,
 		Rdns: sql.NullString{
 			String: hi.Rdns,
 			Valid:  true},

src/models/ip.go

@@ -8,13 +8,14 @@ import (
 	"io"
 	"log"
 	"net/http"
+	"sync"
 	"time"
 
 	"git.paulbsd.com/paulbsd/ipbl/src/config"
 	"xorm.io/xorm"
 )
 
-const SCANLIMIT = 10
+const SCANLIMIT = 100
 
 const IPINFO_WS = "https://ip.paulbsd.com"
 
@@ -42,8 +43,6 @@ func (ip *IP) GetOrCreate(session *xorm.Session) (apiip *APIIP, err error) {
 	ip.City.GetOrCreate(session)
 	ip.Country.GetOrCreate(session)
 	ip.AutonomousSystem.GetOrCreate(session)
-	ip.Host.GetOrCreate(session)
-	ip.Src.GetOrCreate(session)
 	session.Commit()
 
 	var tmpip *IP
@@ -132,26 +131,65 @@ func InsertIPBulk(session *xorm.Session, ips *[]IP) (numinsert int64, numupdate
 }
 
 func ScanIP(cfg *config.Config) (err error) {
-	queryclient := http.Client{}
+	var numthreads = 8
 	for {
 		session := cfg.Db.NewSession()
 		orphans := []IP{}
-		err = session.Where("(as_id is null or country_id is null or city_id is null or rdns = '') and updated < now()-'1d'::interval").Asc("updated").Limit(SCANLIMIT).Find(&orphans)
+		err = session.Where(`((as_id IS NULL
+							  OR country_id IS NULL
+							  OR city_id IS NULL
+							  OR rdns = ''
+							  OR rdns IS NULL) AND updated < now()-'1d'::interval) OR 
+							  (as_id IS NULL
+								AND country_id IS NULL
+								AND city_id IS NULL
+								AND rdns IS NULL)
+							`).Desc("updated").Limit(SCANLIMIT).Find(&orphans)
 		session.Close()
 
 		if err == nil && len(orphans) > 0 {
+			orphanchan := make(chan IP)
+			done := make(chan bool)
+			var wg sync.WaitGroup
+
+			for thr := range make([]int, numthreads) {
+				go ScanOrphan(&wg, orphanchan, done, thr, cfg)
+			}
+
+			for _, orphan := range orphans {
+				orphanchan <- orphan
+			}
+
+			close(orphanchan)
+			<-done
+			wg.Wait()
+
+		} else {
+			time.Sleep(30 * time.Second)
+		}
+	}
+}
+
+func ScanOrphan(wg *sync.WaitGroup, orphans chan IP, done chan bool, thr int, cfg *config.Config) (err error) {
+	wg.Add(1)
+
+	for {
+		orphan, more := <-orphans
+		if more {
 			session := cfg.Db.NewSession()
 			if err != nil {
 				log.Println(err)
 			}
-			for _, orphan := range orphans {
+			queryclient := http.Client{}
+			var query QueryIP
 			query, err := QueryInfo(&queryclient, orphan.IP)
 			if err != nil {
 				log.Println(err)
 				continue
 			}
-				as := &AutonomousSystem{ASID: query.APIAS.ASID, ASName: query.APIAS.ASName}
-				orphan.AutonomousSystem = as
+
+			var as = AutonomousSystem{ASID: query.APIAS.Number, ASName: query.APIAS.Org}
+			orphan.AutonomousSystem = &as
 
 			if query.APICity != "" {
 				var city = City{CityName: query.APICity}
@@ -167,7 +205,6 @@ func ScanIP(cfg *config.Config) (err error) {
 			log.Printf("%s -> \"%s\"\n", orphan.IP, query.Rdns)
 
 			orphan.GetOrCreate(session)
-			}
 
 			err = session.Commit()
 			if err != nil {
@@ -178,9 +215,11 @@ func ScanIP(cfg *config.Config) (err error) {
 				log.Println(err)
 			}
 		} else {
-			time.Sleep(2 * time.Minute)
+			//fmt.Printf("All orphan migrated on thread num %d\n", thr)
+			wg.Done()
+			done <- true
+			return nil
 		}
-		time.Sleep(1 * time.Second)
 	}
 }
 
@@ -261,9 +300,6 @@ type IP struct {
 	AutonomousSystem *AutonomousSystem `xorm:"as_id int index default null"`
 	City             *City             `xorm:"city_id int index default null"`
 	Country          *Country          `xorm:"country_id int index default null"`
-	Src              *Src              `xorm:"src_id int index default null"`
-	Host             *Host             `xorm:"host_id int index default null"`
-	Whois            string            `xorm:"text index default null"`
 	Created          time.Time         `xorm:"created notnull"`
 	Updated          time.Time         `xorm:"updated notnull"`
 }
@@ -280,12 +316,12 @@ type APIIP struct {
 type QueryIP struct {
 	IP         string                `json:"ip"`
 	Rdns       string                `json:"hostname"`
-	APIAS      QueryAutonomousSystem `json:"asn"`
+	APIAS      QueryAutonomousSystem `json:"as"`
 	APICity    string                `json:"city"`
 	APICountry string                `json:"country"`
 }
 
 type QueryAutonomousSystem struct {
-	ASID   int    `json:"id"`
-	ASName string `json:"name"`
+	Number int    `json:"number"`
+	Org    string `json:"org"`
 }

src/models/src.go

@@ -1,6 +1,8 @@
 package models
 
 import (
+	"time"
+
 	"xorm.io/xorm"
 )
 
@@ -46,6 +48,8 @@ func (src *Src) APIParse(apisrc APISrc) (err error) {
 type Src struct {
 	ID      int       `xorm:"pk autoincr"`
 	Src     string    `xorm:"text unique(srcindex) src" json:"src"`
+	Created time.Time `xorm:"created notnull"`
+	Updated time.Time `xorm:"updated notnull"`
 }
 
 type APISrc struct {

src/routers/funcs.go

@@ -22,6 +22,10 @@ func RegisterRoutes(e *echo.Echo, ctx *context.Context, cfg *config.Config) {
 		</html>`)
 	})
 
+	e.GET("/health", func(c echo.Context) error {
+		return c.HTML(http.StatusOK, `OK`)
+	})
+
 	e.GET("/ip/:ip", func(c echo.Context) (err error) {
 		session := cfg.Db.NewSession()
 		defer session.Close()