Merge branch 'develop' into master for 1.11.1

Changes since last version:
* Fix detection of BIRD v2.x.y
* Fix birdc command in RoutesFiltered
* Use worker-threads to parse in parallel.
  This speeds up parsing of large responses e.g. BGP full-table.
* Add flag "worker-pool-size" to control number of threads while parsing
* Configuration: add setting for ttl value to control caching of bird responses
* Configuration: change default location to /etc/birdwatcher

CHANGELOG

@@ -1,4 +1,13 @@
 
+1.11.1
+
+* Fix detection of BIRD v2.x.y
+* Fix birdc command in RoutesFiltered
+* Use worker-threads to parse in parallel. This speeds up parsing of large responses e.g. BGP full-table.
+* Add flag "worker-pool-size" to control number of threads while parsing
+* Configuration: add setting for ttl value to control caching of bird responses
+* Configuration: change default location to /etc/birdwatcher
+
 1.11.0
 
 * Parser: support BIRD v2.x with multiprotocol BGP and channels

VERSION

@@ -1 +1 @@
-1.11.0
+1.11.1

bird/bird.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"io"
 	"reflect"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -48,7 +49,11 @@ func fromCache(key string) (Parsed, bool) {
 }
 
 func toCache(key string, val Parsed) {
-	val["ttl"] = time.Now().Add(5 * time.Minute)
+	var ttl int = 5
+	if ClientConf.CacheTtl > 0 {
+		ttl = ClientConf.CacheTtl
+	}
+	val["ttl"] = time.Now().Add(time.Duration(ttl) * time.Minute)
 	Cache.Lock()
 	Cache.m[key] = val
 	Cache.Unlock()
@@ -199,7 +204,7 @@ func RoutesProtoCount(protocol string) (Parsed, bool) {
 }
 
 func RoutesFiltered(protocol string) (Parsed, bool) {
-	cmd := routeQueryForChannel("route all filtered " + protocol)
+	cmd := routeQueryForChannel("route all filtered protocol " + protocol)
 	return RunAndParse(cmd, parseRoutes)
 }
 
@@ -326,7 +331,8 @@ func routeQueryForChannel(cmd string) string {
 		return cmd
 	}
 
-	if len(version) == 0 || int(version[0]) < 2 {
+	v, err := strconv.Atoi(string(version[0]))
+	if err != nil || v <= 2 {
 		return cmd
 	}
 

bird/config.go

@@ -13,6 +13,7 @@ type BirdConfig struct {
 	Listen         string
 	ConfigFilename string `toml:"config"`
 	BirdCmd        string `toml:"birdc"`
+	CacheTtl       int    `toml:"ttl"`
 }
 
 type ParserConfig struct {

bird/parser.go

@@ -6,13 +6,16 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 )
 
+// WorkerPoolSize is the number of go routines used to parse routing tables concurrently
+var WorkerPoolSize = 8
+
 var (
 	ParserConf ParserConfig
 	regex      struct {
-		lineSeperator *regexp.Regexp
-		status        struct {
+		status struct {
 			startLine     *regexp.Regexp
 			routerID      *regexp.Regexp
 			currentServer *regexp.Regexp
@@ -177,15 +180,107 @@ func parseSymbols(reader io.Reader) Parsed {
 	return Parsed{"symbols": res}
 }
 
-func parseRoutes(reader io.Reader) Parsed {
-	res := Parsed{}
-	routes := []Parsed{}
-	route := Parsed{}
+type blockJob struct {
+	lines    []string
+	position int
+}
 
+type blockParsed struct {
+	items    []Parsed
+	position int
+}
+
+func parseRoutes(reader io.Reader) Parsed {
+	jobs := make(chan blockJob)
+	out := startRouteWorkers(jobs)
+
+	res := startRouteConsumer(out)
+	defer close(res)
+
+	pos := 0
+	block := []string{}
 	lines := newLineIterator(reader, true)
+
 	for lines.next() {
 		line := lines.string()
 
+		if line[0] != 32 && line[0] != 9 && len(block) > 0 {
+			jobs <- blockJob{block, pos}
+			pos++
+			block = []string{}
+		}
+
+		block = append(block, line)
+	}
+
+	if len(block) > 0 {
+		jobs <- blockJob{block, pos}
+	}
+
+	close(jobs)
+
+	return <-res
+}
+
+func startRouteWorkers(jobs chan blockJob) chan blockParsed {
+	out := make(chan blockParsed)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(WorkerPoolSize)
+	go func() {
+		for i := 0; i < WorkerPoolSize; i++ {
+			go workerForRouteBlockParsing(jobs, out, wg)
+		}
+		wg.Wait()
+		close(out)
+	}()
+
+	return out
+}
+
+func startRouteConsumer(out <-chan blockParsed) chan Parsed {
+	res := make(chan Parsed)
+
+	go func() {
+		byBlock := map[int][]Parsed{}
+		count := 0
+		for r := range out {
+			count++
+			byBlock[r.position] = r.items
+		}
+		res <- Parsed{"routes": sortedSliceForRouteBlocks(byBlock, count)}
+	}()
+
+	return res
+}
+
+func sortedSliceForRouteBlocks(byBlock map[int][]Parsed, numBlocks int) []Parsed {
+	res := []Parsed{}
+
+	for i := 0; i < numBlocks; i++ {
+		routes, ok := byBlock[i]
+		if !ok {
+			continue
+		}
+
+		res = append(res, routes...)
+	}
+
+	return res
+}
+
+func workerForRouteBlockParsing(jobs <-chan blockJob, out chan<- blockParsed, wg *sync.WaitGroup) {
+	for j := range jobs {
+		parseRouteLines(j.lines, j.position, out)
+	}
+	wg.Done()
+}
+
+func parseRouteLines(lines []string, position int, ch chan<- blockParsed) {
+	route := Parsed{}
+	routes := []Parsed{}
+
+	for _, line := range lines {
 		if specialLine(line) {
 			continue
 		}
@@ -233,8 +328,7 @@ func parseRoutes(reader io.Reader) Parsed {
 		routes = append(routes, route)
 	}
 
-	res["routes"] = routes
-	return res
+	ch <- blockParsed{routes, position}
 }
 
 func parseMainRouteDetail(groups []string, route Parsed) {

birdwatcher.go

@@ -13,7 +13,7 @@ import (
 )
 
 //go:generate versionize
-var VERSION = "1.11.0"
+var VERSION = "1.11.1"
 
 func isModuleEnabled(module string, modulesEnabled []string) bool {
 	for _, enabled := range modulesEnabled {
@@ -89,6 +89,7 @@ func PrintServiceInfo(conf *Config, birdConf bird.BirdConfig) {
 	log.Println("Starting Birdwatcher")
 	log.Println("            Using:", birdConf.BirdCmd)
 	log.Println("           Listen:", birdConf.Listen)
+	log.Println("        Cache TTL:", birdConf.CacheTtl)
 
 	// Endpoint Info
 	if len(conf.Server.AllowFrom) == 0 {
@@ -107,9 +108,12 @@ func PrintServiceInfo(conf *Config, birdConf bird.BirdConfig) {
 
 func main() {
 	bird6 := flag.Bool("6", false, "Use bird6 instead of bird")
-	configfile := flag.String("config", "./etc/ecix/birdwatcher.conf", "Configuration file location")
+	workerPoolSize := flag.Int("worker-pool-size", 8, "Number of go routines used to parse routing tables concurrently")
+	configfile := flag.String("config", "etc/birdwatcher/birdwatcher.conf", "Configuration file location")
 	flag.Parse()
 
+	bird.WorkerPoolSize = *workerPoolSize
+
 	endpoints.VERSION = VERSION
 	bird.InstallRateLimitReset()
 	// Load configurations

etc/ecix/birdwatcher.conf

@@ -45,12 +45,13 @@ requests_per_minute = 10
 listen = "0.0.0.0:29188"
 config = "/etc/bird.conf"
 birdc  = "/sbin/birdc"
-
+ttl = 5 # time to live (in minutes) for caching of cli output
 
 [bird6]
 listen = "0.0.0.0:29189"
 config = "/etc/bird6.conf"
 birdc  = "/sbin/birdc6"
+ttl = 5 # time to live (in minutes) for caching of cli output
 
 [parser]
 # Remove fields e.g. interface
@@ -60,4 +61,3 @@ filter_fields = []
 per_peer_tables = true
 peer_protocol_prefix = 'ID'
 pipe_protocol_prefix = 'P'
-