Multiple changes

Fixes #6
Fixes #7

asn2asd.go

@@ -26,19 +26,18 @@ func (i *ASN2ASDescClient) importedDates() ([]string, error) {
 }
 
 // Current returns the latest known result for an IP2ASN lookup.
-func (i *ASN2ASDescClient) Current(ASN int) string {
+func (i *ASN2ASDescClient) Current(ASN int) (string, error) {
 	allDates, err := i.importedDates()
 	if err != nil {
-		return ""
+		return "", err
 	}
 	if len(allDates) < 0 {
-		return ""
+		return "", err
 	}
 	current := allDates[len(allDates)-1]
 	result, err := redis.String(i.conn.Do("HGET", fmt.Sprintf("asd:%d", ASN), current))
 	if err != nil {
-		return ""
+		return "", err
 	}
-
-	return result
+	return result, nil
 }

bgpdump.go

@@ -2,6 +2,7 @@ package internet
 
 import (
 	"bufio"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -11,30 +12,26 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/garyburd/redigo/redis"
 )
 
 // RefreshBGPDump ensures that the latest dump available is the one which is installed.
-func RefreshBGPDump(conn redis.Conn) error {
+func RefreshBGPDump(conn redis.Conn) (int, error) {
 	for _, b := range []BGPDump{
 		{Date: time.Now()},
 		{Date: time.Now().Add(-time.Duration(time.Hour * 24))},
 	} {
 		err := b.Download()
 		if err != nil {
-			return err
+			return 0, err
 		}
 		if b.IsDownloaded() {
-			if err := b.Import(conn); err != nil {
-				return err
-			}
-			return nil
+			return b.Import(conn)
 		}
 	}
-	return nil
+	return 0, nil
 }
 
 // BGPDump encapuslates downloading and importing of BGP dumps.
@@ -43,35 +40,60 @@ type BGPDump struct {
 }
 
 // Import stores the contents of a downloaded BGP dump into a redis server.
-func (b *BGPDump) Import(conn redis.Conn) error {
-
+// -1 is returned if the dump is alredy imported into redis.
+func (b *BGPDump) Import(conn redis.Conn) (int, error) {
 	alreadyImported, err := redis.Bool(conn.Do("SISMEMBER", "i2a:imported_dates", b.day()))
 	if err != nil {
-		panic(err)
+		return 0, err
 	}
 	if alreadyImported {
-		return nil
+		return -1, nil
 	}
 	c := exec.Command("bgpdump", "-m", b.Path())
-
 	stdout, err := c.StdoutPipe()
 	if err != nil {
-		return err
+		return 0, err
+	}
 
+	type nErr struct {
+		n   int
+		err error
 	}
-	var wg sync.WaitGroup
-	wg.Add(1)
+
+	parseC := make(chan nErr)
 	go func(r io.Reader) {
-		defer wg.Done()
-		b.parseBGPCSV(r, conn)
+		defer func() {
+			if err := recover(); err != nil {
+				log.Println(err)
+				switch err.(type) {
+				case error:
+					parseC <- nErr{
+						err: err.(error),
+					}
+				default:
+					parseC <- nErr{err: errors.New("unknown error")}
+				}
+			}
+		}()
+		n, err := b.parseBGPCSV(r, conn)
+		parseC <- nErr{n, err}
 	}(stdout)
-	err = c.Run()
-	if err != nil {
-		return err
+
+	execC := make(chan error)
+	go func() {
+		err = c.Run()
+		if err != nil {
+			execC <- err
+		}
+	}()
+
+	select {
+	case err := <-execC:
+		return 0, err
+	case ne := <-parseC:
+		return ne.n, ne.err
 	}
-	wg.Wait()
 
-	return nil
 }
 
 // IsDownloaded returns true if the BGPDump archive is downloaded locally.
@@ -152,18 +174,20 @@ func (b *BGPDump) Download() error {
 
 }
 
-func (b *BGPDump) parseBGPCSV(r io.Reader, conn redis.Conn) error {
+func (b *BGPDump) parseBGPCSV(r io.Reader, conn redis.Conn) (int, error) {
 	day := b.day()
-
-	start := time.Now()
 	s := bufio.NewScanner(r)
 	n := 0
 	var asn string
 	for s.Scan() {
 		cols := strings.Split(s.Text(), "|")
 		if len(cols) < 7 {
-			log.Printf("Too few columns in %s:%d: %s", filepath.Base(b.Path()), n, s.Text())
-			continue
+			return n, ParseError{
+				Message: "too few columns",
+				Path:    filepath.Base(b.Path()),
+				LineNum: n,
+				Line:    s.Text(),
+			}
 		}
 		block := cols[5]
 
@@ -174,8 +198,12 @@ func (b *BGPDump) parseBGPCSV(r io.Reader, conn redis.Conn) error {
 			asns := strings.Split(asPath, " ")
 			asn = asns[len(asns)-1]
 			if asn == "" {
-				log.Printf("No ASPATH data for %s:%d: %s", filepath.Base(b.Path()), n, s.Text())
-				continue
+				return n, ParseError{
+					Message: "no ASPATH data",
+					Path:    filepath.Base(b.Path()),
+					LineNum: n,
+					Line:    s.Text(),
+				}
 			}
 		}
 		conn.Send("HSET", fmt.Sprintf("i2a:%s", block), day, asn)
@@ -183,20 +211,16 @@ func (b *BGPDump) parseBGPCSV(r io.Reader, conn redis.Conn) error {
 		if n%10000 == 0 {
 			err := conn.Flush()
 			if err != nil {
-				panic(err)
+				return 0, err
 			}
 		}
 	}
 	conn.Send("SADD", "i2a:imported_dates", day)
 	err := conn.Flush()
 	if err != nil {
-		panic(err)
+		return 0, err
 	}
-
-	log.Printf("Imported %d rows from %s in %s",
-		n, filepath.Base(b.Path()), time.Since(start))
-
-	return nil
+	return n, nil
 }
 
 // Path returns the absolute path to the target archive dump download file.

bgpdump_test.go

@@ -3,7 +3,6 @@ package internet
 import (
 	"os"
 	"testing"
-
 	"time"
 
 	"github.com/garyburd/redigo/redis"
@@ -26,8 +25,13 @@ func TestParseDump(t *testing.T) {
 	}
 	conn := redigomock.NewConn()
 	b := BGPDump{Date: time.Date(2015, 1, 1, 0, 0, 0, 0, time.UTC)}
-	b.parseBGPCSV(file, conn)
-
+	n, err := b.parseBGPCSV(file, conn)
+	if err != nil {
+		panic(err)
+	}
+	if n != 801 {
+		t.Fatalf("expected 801 imported entries, got %d", n)
+	}
 }
 
 func TestParseBrokenDump(t *testing.T) {
@@ -35,12 +39,16 @@ func TestParseBrokenDump(t *testing.T) {
 	if err != nil {
 		panic(err)
 	}
-
 	conn := redigomock.NewConn()
 	b := BGPDump{Date: time.Date(2015, 1, 2, 0, 0, 0, 0, time.UTC)}
-	err = b.parseBGPCSV(file, conn)
-	if err != nil {
-		panic(err)
+	n, err := b.parseBGPCSV(file, conn)
+	if err == nil {
+		t.Fatalf("expected parse error")
+	}
+	if n != 5 {
+		t.Fatalf("expected parse error on line 5 but %d lines were read", n)
+	}
+	if _, ok := err.(ParseError); !ok {
+		t.Fatalf("expected parse error, got %s", err)
 	}
-
 }

cidrreport.go

@@ -1,7 +1,6 @@
 package internet
 
 import (
-	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -53,39 +52,42 @@ func (b *CIDRReport) IsDownloaded() bool {
 }
 
 // Import stores the contents of a downloaded BGP dump into a redis server.
-func (b *CIDRReport) Import(conn redis.Conn) error {
+// -1 is returned if the dump is alredy imported into redis.
+func (b *CIDRReport) Import(conn redis.Conn) (int, error) {
 
 	alreadyImported, err := redis.Bool(conn.Do("SISMEMBER", "asd:imported_dates", b.day()))
 	if err != nil {
-		return err
+		return 0, err
 	}
 	if alreadyImported {
-		return nil
+		return -1, nil
 	}
 
 	file, err := os.Open(b.Path())
 	if err != nil {
-		return err
+		return 0, err
 	}
 	n := 0
 	day := b.day()
-	err = parseReport(file, func(asd *ASDescription) {
+	err = parseReport(file, func(asd *ASDescription) error {
 		conn.Send("HSET", fmt.Sprintf("asd:%d", asd.ASN), day,
 			fmt.Sprintf("%s, %s", asd.Description, asd.CountryCode))
 		n++
 		if n%10000 == 0 {
 			err := conn.Flush()
 			if err != nil {
-				panic(err)
+				return err
+
 			}
 		}
+		return nil
 	})
 	conn.Send("SADD", "asd:imported_dates", day)
 	err = conn.Flush()
 	if err != nil {
-		panic(err)
+		return 0, err
 	}
-	return nil
+	return n, nil
 }
 
 // Download fetches http://www.cidr-report.org/as2.0/autnums.html and stores it
@@ -134,26 +136,23 @@ func (b *CIDRReport) Download() error {
 }
 
 // RefreshCIDRReport ensures that the latest dump available is the one which is installed.
-func RefreshCIDRReport(conn redis.Conn) error {
+func RefreshCIDRReport(conn redis.Conn) (int, error) {
 	for _, b := range []CIDRReport{
 		{Date: time.Now()},
 		{Date: time.Now().Add(-time.Duration(time.Hour * 24))},
 	} {
 		err := b.Download()
 		if err != nil {
-			return err
+			return 0, err
 		}
 		if b.IsDownloaded() {
-			if err := b.Import(conn); err != nil {
-				return err
-			}
-			return nil
+			return b.Import(conn)
 		}
 	}
-	return nil
+	return 0, nil
 }
 
-func parseReport(r io.Reader, emitter func(*ASDescription)) error {
+func parseReport(r io.Reader, emitter func(*ASDescription) error) error {
 	z := html.NewTokenizer(r)
 	n := 0
 	depth := 0
@@ -169,7 +168,13 @@ loop:
 				desc := strings.TrimSpace(string(z.Text()))
 				ccpos := strings.LastIndex(desc, ",")
 				if ccpos == -1 {
-					return fmt.Errorf("Could not parse country code from %d %s", asn, desc)
+					return ParseError{
+						Message: "Could not parse country code",
+						Path:    "cidrreport",
+						LineNum: n,
+						Line:    fmt.Sprintf("asn:%s desc:%s", asn, desc),
+					}
+
 				}
 				emitter(&ASDescription{
 					ASN:         *asn,
@@ -183,7 +188,11 @@ loop:
 				var err error
 				i, err := strconv.Atoi(strings.TrimSpace(asnstr))
 				if err != nil {
-					return err
+					return ParseError{
+						Message: err.Error(),
+						Path:    "cidrreport",
+						LineNum: n,
+					}
 				}
 				asn = &i
 			}
@@ -199,7 +208,12 @@ loop:
 		}
 	}
 	if n == 0 {
-		return errors.New("No entries found, the parsing failed")
+		return ParseError{
+			Message: "no entries found",
+			Path:    "cidrreport",
+			LineNum: n,
+		}
+
 	}
 	return nil
 }

cidrreport_test.go

@@ -8,8 +8,9 @@ import (
 func TestParseCidrreport(t *testing.T) {
 	file, err := os.Open("testdata/autnums.sample.txt")
 	n := 0
-	err = parseReport(file, func(*ASDescription) {
+	err = parseReport(file, func(*ASDescription) error {
 		n++
+		return nil
 	})
 	if err != nil {
 		t.Error(err)
@@ -21,8 +22,17 @@ func TestParseCidrreport(t *testing.T) {
 
 func TestParseBrokenCidrreport(t *testing.T) {
 	file, err := os.Open("testdata/autnums.invalid.sample.txt")
-	err = parseReport(file, func(*ASDescription) {})
-	if err == nil {
-		t.Errorf("Parse error was expected")
+	err = parseReport(file, func(*ASDescription) error { return nil })
+	if _, ok := err.(ParseError); !ok {
+		t.Fatalf("expected parse error, got %s", err)
+	}
+}
+
+func TestParseBrokenLineCidrreport(t *testing.T) {
+	file, err := os.Open("testdata/autnums.invalid.sample2.txt")
+	err = parseReport(file, func(*ASDescription) error { return nil })
+
+	if _, ok := err.(ParseError); !ok {
+		t.Fatalf("expected parse error, got %s", err)
 	}
 }

cmd/dev/dev.go

@@ -18,7 +18,6 @@ import (
 func main() {
 	flag.Parse()
 	pool = newPool(*redisServer, "")
-
 	mainCidrr()
 	mainIP2ASN()
 }
@@ -27,6 +26,7 @@ func mainCidrr() {
 	cc := internet.CIDRReport{
 		Date: time.Now(),
 	}
+	log.Printf("DOWNLOAD %v", cc)
 	err := cc.Download()
 	if err != nil {
 		panic(err)
@@ -35,11 +35,14 @@ func mainCidrr() {
 	if cc.IsDownloaded() {
 		conn := pool.Get()
 		defer conn.Close()
-		err := cc.Import(conn)
+		log.Printf("IMPORT %v", cc)
+		start := time.Now()
+		n, err := cc.Import(conn)
 		if err != nil {
 			panic(err)
 		}
-
+		log.Printf("Imported %d rows from %s in %s",
+			n, filepath.Base(cc.Path()), time.Since(start))
 	}
 }
 
@@ -58,13 +61,25 @@ func DoLookup() {
 	q := internet.NewIP2ASNClient(conn)
 	q2 := internet.NewASN2ASDescClient(conn)
 	for _, i := range []string{"8.8.8.8", "5.150.255.150", "127.0.0.1"} {
-		res := q.Current(i)
+		res, err := q.Current(i)
+		if err != nil {
+			panic(err)
+		}
 		log.Printf("current   : %s: %s ", i, res)
 		if res != nil {
-			log.Printf("desc      : %s", q2.Current(res.ASN))
+			cur, err := q2.Current(res.ASN)
+			if err != nil {
+				panic(err)
+			}
+			log.Printf("desc      : %s", cur)
 		}
 		log.Printf("allhistory: %s", i)
-		for _, v := range q.AllHistory(i) {
+		allhist, err := q.AllHistory(i)
+		if err != nil {
+			panic(err)
+		}
+
+		for _, v := range allhist {
 			log.Println(v.String())
 		}
 	}
@@ -91,20 +106,23 @@ func DoIndex() {
 					log.Println("Recovered in f", r, b.Path())
 				}
 			}()
-			// log.Printf("DOWNLOAD %s", b.Path())
+			log.Printf("DOWNLOAD %s", b.Path())
 			err := b.Download()
 			if err != nil {
 				panic(err)
 			}
 			if b.IsDownloaded() {
-				// log.Printf("IMPORT %s", b.Path())
+				log.Printf("IMPORT %s", b.Path())
 				conn := pool.Get()
 				defer conn.Close()
-				err := b.Import(conn)
-
+				start := time.Now()
+				n, err := b.Import(conn)
 				if err != nil {
 					panic(err)
 				}
+				log.Printf("Imported %d rows from %s in %s",
+					n, filepath.Base(b.Path()), time.Since(start))
+
 			}
 		}(b)
 	}

internet.go

@@ -37,10 +37,22 @@ https://github.com/CIRCL/ASN-Description-History.
 package internet
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
 )
 
+type ParseError struct {
+	Message string
+	Path    string
+	LineNum int
+	Line    string
+}
+
+func (p ParseError) Error() string {
+	return fmt.Sprintf("parse error: %s in %s:%d: %s", p.Message, p.Path, p.LineNum, p.Line)
+}
+
 var dataDir = filepath.Join(os.TempDir(), "internet")
 
 // SetDataDir sets the storage directory for downloads cache and temporary files.

ip2asn.go

@@ -32,41 +32,44 @@ type IP2ASNClient struct {
 }
 
 // Current returns the latest known result for an IP2ASN lookup.
-func (i *IP2ASNClient) Current(IP string) *ASNResult {
+func (i *IP2ASNClient) Current(IP string) (*ASNResult, error) {
 	ip, err := i.parseIP(IP)
 	if err != nil {
-		return &ASNResult{}
+		return &ASNResult{}, err
 	}
 	var current string
 	allDates, err := i.importedDates()
 	if err != nil {
-		return &ASNResult{}
+		return &ASNResult{}, err
 	}
 
 	if len(allDates) < 0 {
-		return &ASNResult{}
+		return &ASNResult{}, err
 	}
 	current = allDates[len(allDates)-1]
 
-	results := i.dates(ip, []string{current})
+	results, err := i.dates(ip, []string{current})
+	if err != nil {
+		return nil, err
+	}
+
 	if len(results) > 0 {
-		return &results[0]
+		return &results[0], nil
 	}
-	return nil
+	return nil, nil
 }
 
 // AllHistory returns the full history for the given IP address.
-func (i *IP2ASNClient) AllHistory(IP string) []ASNResult {
+func (i *IP2ASNClient) AllHistory(IP string) ([]ASNResult, error) {
 	ip, err := i.parseIP(IP)
 	if err != nil {
-		return []ASNResult{}
+		return []ASNResult{}, err
 	}
 	dates, err := i.importedDates()
 	if err != nil {
-		panic(err)
+		return []ASNResult{}, err
 	}
-	result := i.dates(ip, dates)
-	return result
+	return i.dates(ip, dates)
 }
 
 func (i *IP2ASNClient) parseIP(IP string) (net.IP, error) {
@@ -103,7 +106,7 @@ func (i *IP2ASNClient) importedDates() ([]string, error) {
 }
 
 // dates resolves IP2ASN for all date entries, if available.
-func (i *IP2ASNClient) dates(IP net.IP, dates []string) []ASNResult {
+func (i *IP2ASNClient) dates(IP net.IP, dates []string) ([]ASNResult, error) {
 	keys := i.keys(IP)
 	for _, d := range dates {
 		for _, k := range keys {
@@ -123,14 +126,14 @@ func (i *IP2ASNClient) dates(IP net.IP, dates []string) []ASNResult {
 					if err == redis.ErrNil {
 						continue
 					}
-					panic(err)
+					return []ASNResult{}, err
 				}
 
 				timedate, err := time.Parse("20060102", date)
 				asn, err := strconv.Atoi(r)
 				if err != nil {
 					// redis data error
-					panic(err)
+					return []ASNResult{}, err
 				}
 				results = append(results, ASNResult{
 					Mask: keys[idx],
@@ -144,7 +147,7 @@ func (i *IP2ASNClient) dates(IP net.IP, dates []string) []ASNResult {
 		}
 
 	}
-	return results
+	return results, nil
 }
 
 // RIPE-NCC-RIS BGP IPv6 Anchor Prefix @RRC00