go-bgp

archive.go (13840B)

---

 package archive
 
 import (
 	"fmt"
 	"log"
 	//"io/ioutil"
 	//"bytes"
 	"bufio"
 	"compress/bzip2"
 	"errors"
 	mrt "go-bgp/mrt"
 	"io"
 	"net/url"
 	"os"
 	"path/filepath"
 	"sort"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 	"unicode"
 )
 
 //constants for archentryfile.ftype
 const (
 	unknownfile = iota
 	mrtfile
 	xmlfile
 	ftypesize
 )
 
 var (
 	ftypestrings = [ftypesize]string{
 		"UNKNOWN",
 		"MRT",
 		"XML",
 	}
 	errbadft     = errors.New("error:  unknown file type")
 	errparsedate = errors.New("error: parsing date")
 	errbadscan   = errors.New("bufio scanner failed")
 )
 
 type reply struct {
 	data []byte
 	err  error
 }
 
 //To perform a query asynchronously on possibly many files we fire multiple goroutines
 //that all write their results to chan reply, and we also need the waitgroup
 //to know when we should close the channel to end the http transaction
 type archive interface {
 	Query(queryParams)
 	visit(string, os.FileInfo, error) error
 }
 
 type queryParams struct {
 	from  time.Time
 	to    time.Time
 	rtype int
 	rch   chan reply
 	wg    *sync.WaitGroup
 }
 
 type xmlstring struct {
 	timestr string
 	msg     string
 	time    time.Time
 }
 
 func (x *xmlstring) String() string {
 	return x.msg
 }
 
 //implements Sort interface by time.Time
 type archentryfile struct {
 	path  string
 	sdate time.Time
 	sz    int64
 	ftype int
 }
 
 type timeentryslice []archentryfile
 
 func (p timeentryslice) Len() int {
 	return len(p)
 }
 
 func (p timeentryslice) Less(i, j int) bool {
 	return p[i].sdate.Before(p[j].sdate)
 }
 
 func (p timeentryslice) Swap(i, j int) {
 	p[i], p[j] = p[j], p[i]
 }
 
 type fsarchive struct {
 	rootpathstr    string
 	entryfiles     *timeentryslice
 	tempentryfiles timeentryslice //once we finish operating on that, we change the entryfiles pointer
 	curyr          int            //in case we need to walk dirs to find dates,these keep state
 	curmon         int
 	curday         int
 	reqchan        chan string
 	scanning       bool
 	Scanwg         *sync.WaitGroup // expose it so callers are able to wait for scan to finish
 	scanch         chan struct{}
 	timedelta      time.Duration
 	descriminator  string //a string to be matched if a file is to be added while visiting the fs hier
 	Conf           *fsarconf
 	//present tha archive as a restful resource
 	PutNotAllowed
 	PostNotAllowed
 	DeleteNotAllowed
 }
 
 type fsarconf struct {
 	ar *fsarchive
 	PutNotAllowed
 	PostNotAllowed
 	DeleteNotAllowed
 }
 
 //in order not to block in the following GETs, we need to
 //fire a new goroutine to send the reply on the channel
 // the reason is that we create the channel here and we must
 //return it to the responsewriter and any sends would block
 //without the receiver being ready.
 func (fsc *fsarconf) Get(values url.Values) (int, chan reply) {
 	retc := make(chan reply)
 	go func() {
 		defer close(retc) //must close the chan to let the listener finish.
 		if fsc.ar == nil {
 			log.Printf("nil archive in fsarconf. ignoring request\n")
 			return
 		}
 		if _, ok := values["range"]; ok {
 			if len(*fsc.ar.entryfiles) > 0 {
 				f := *fsc.ar.entryfiles
 				dates := fmt.Sprintf("[From:%s - To:%s . Delta:%s]\n", f[0].sdate, f[len(f)-1].sdate, fsc.ar.timedelta)
 				retc <- reply{data: []byte(dates), err: nil}
 				return
 			}
 			retc <- reply{data: nil, err: errempty}
 			return
 		}
 		if _, ok := values["files"]; ok {
 			for _, f := range *fsc.ar.entryfiles {
 				retc <- reply{data: []byte(fmt.Sprintf("[Filename:%s .Type:%s]\n", filepath.Base(f.path), ftypestrings[f.ftype])), err: nil}
 			}
 			return
 		}
 		return
 	}()
 	return 200, retc
 }
 
 func (fsa *fsarchive) Get(values url.Values) (int, chan reply) {
 	var (
 		grwg  sync.WaitGroup
 		ftype int
 	)
 	retc := make(chan reply)
 	timeAstrs, ok1 := values["start"]
 	timeBstrs, ok2 := values["end"]
 	tpstr, ok3 := values["type"]
 	if len(timeAstrs) != len(timeBstrs) || !ok1 || !ok2 {
 		retc <- reply{data: nil, err: errbadreq}
 		goto done
 	}
 	if !ok3 { //no type specified. ensuring it's 0 (unknownfile)
 		ftype = unknownfile
 	} else {
 		for tpi, tps := range ftypestrings {
 			//XXX: here i only respect the first type argument
 			if tps == tpstr[0] {
 				ftype = tpi
 				break
 			}
 			if tpi == ftypesize-1 { //we didn't find it let the user know
 				log.Printf("unknown return type %s in request", tpstr[0])
 				retc <- reply{data: nil, err: errbadtype}
 				goto done
 			}
 		}
 	}
 	//This will parse pairs of start=YYYYMMDDHHMMSS&end=YYYYMMDDHHMMSS requests
 	for i := 0; i < len(timeAstrs); i++ {
 		log.Printf("timeAstr:%s timeBstr:%s", timeAstrs[i], timeBstrs[i])
 		timeA, errtime := time.Parse("20060102150405", timeAstrs[i])
 		timeB, errtime := time.Parse("20060102150405", timeBstrs[i])
 		if errtime != nil || timeB.Before(timeA) {
 			retc <- reply{data: nil, err: errbaddate}
 		} else {
 			//buf.WriteString(fmt.Sprintf("quering from t0:%s - t1:%s\n", timeA, timeB))
 			qp := queryParams{from: timeA, to: timeB, rtype: ftype, rch: retc, wg: &grwg}
 			fsa.Query(qp) //this will fire a new goroutine
 		}
 	}
 	// the last goroutine that will wait for all we invoked and close the chan
 	go func(wg *sync.WaitGroup) {
 		wg.Wait()   //wait for all the goroutines to finish sending
 		close(retc) //close the chan so that range in responsewriter will finish
 		log.Printf("closing the chan\n")
 	}(&grwg)
 done:
 	return 200, retc
 }
 
 //func (fsar *fsarchive) Query(ta, tb time.Time, retc chan reply, wg *sync.WaitGroup) {
 func (fsar *fsarchive) Query(qp queryParams) {
 	ta := qp.from
 	tb := qp.to
 	retc := qp.rch
 	rt := qp.rtype
 	wg := qp.wg
 	log.Printf("querying from %s to %s return as:%s \n", ta, tb, ftypestrings[rt])
 	go func(rc chan<- reply) {
 		wg.Add(1)
 		ef := *fsar.entryfiles
 		var (
 			scanner *bufio.Scanner
 			date    time.Time
 			err     error
 		)
 		defer wg.Done()
 		//case no files in archive
 		if len(ef) == 0 {
 			rc <- reply{nil, errempty}
 			return
 		}
 		//case request not on the served range
 		if tb.Before(ef[0].sdate) || ta.After(ef[len(ef)-1].sdate.Add(fsar.timedelta)) {
 			rc <- reply{nil, errdate}
 			return
 		}
 		//get the start end indices
 		i := sort.Search(len(ef), func(i int) bool {
 			return ef[i].sdate.After(ta.Add(-fsar.timedelta - time.Second))
 		})
 		j := sort.Search(len(ef), func(i int) bool {
 			return ef[i].sdate.After(tb)
 		})
 		for k := i; k < j; k++ {
 			//fext := filepath.Ext(ef[k].path)
 			file, ferr := os.Open(ef[k].path) //not calling defer cause i want to close files in the end of the loop
 			if ferr != nil {
 				log.Println("failed opening file: ", ef[k].path, " with ", ferr)
 				continue
 			}
 			ft := ef[k].ftype
 			scanner, err = getScanner(file, ft)
 			if err != nil {
 				log.Println("failed acquiring bufio scanner on: ", ef[k].path, " ", err)
 				continue
 			}
 
 			startt := time.Now()
 			for scanner.Scan() {
 				b := scanner.Bytes()
 				if len(b) == 0 {
 					log.Printf("0 number of bytes scanned from :%s", ef[k].path)
 				} else {
 					date, err = getDate(b, ft)
 					if err != nil {
 						log.Printf("Query: getting date on file:%s error: %s", ef[k].path, err)
 						continue
 					}
 				}
 				if date.After(ta) && date.Before(tb) {
 					log.Printf("succesfully sending data for date:%s", date)
 					rc <- reply{data: b, err: nil}
 				} else if date.After(tb) {
 					break //XXX: this is correct only on ordered files (increasing dates)
 				}
 			}
 			if err := scanner.Err(); err != nil && err != io.EOF {
 				log.Printf("file scanner error:%s\n", err)
 			}
 			log.Printf("finished parsing file %s size %d in %s\n", ef[k].path, ef[k].sz, time.Since(startt))
 			file.Close()
 		}
 		return
 	}(retc)
 }
 
 func getScanner(f *os.File, ft int) (scanner *bufio.Scanner, err error) {
 	scanner = nil
 	fext := filepath.Ext(f.Name())
 	if fext == ".bz2" {
 		log.Printf("bunzip2 file. opening decompression stream\n")
 		bzreader := bzip2.NewReader(f)
 		scanner = bufio.NewScanner(bzreader)
 	} else {
 		log.Printf("no extension on file: %s. opening normally\n", f.Name())
 		scanner = bufio.NewScanner(f)
 	}
 	if ft == mrtfile {
 		log.Printf("mrt file: %s. setting splitfunc to SplitMrt\n", f.Name())
 		scanner.Split(mrt.SplitMrt)
 	}
 	return
 }
 
 func getDate(data []byte, ftype int) (t time.Time, err error) {
 	switch ftype {
 	case mrtfile:
 		hdr, err := mrt.NewMrtHdr(data[:mrt.MrtHdr_size])
 		if err != nil {
 			log.Printf("getDate: error in creating MRT header:%s", err)
 		} else {
 			t = time.Unix(int64(hdr.Mrt_timestamp), 0)
 		}
 	case xmlfile:
 		str := string(data)
 		dateindi := strings.Index(str, "<DATETIME>")
 		if dateindi == -1 {
 			log.Println("getDate: could not locate DATETIME string in xml msg: ", str)
 			err = errbaddate
 			return
 		}
 		dateindi = dateindi + 10 // go to start of date data
 		dateindj := strings.Index(str[dateindi:], "</DATETIME>")
 		if dateindj == -1 {
 			log.Println("getDate: could not locate closing </DATETIME> string in xml msg: ", str)
 			err = errbaddate
 			return
 		}
 		dateindj = dateindj + dateindi // to return it to the relative start of line pos
 		t, err = time.Parse(time.RFC3339, str[dateindi:dateindj])
 		if err != nil {
 			log.Println("getDate: could not parse datetime: %s\n", err)
 		}
 	default:
 		err = errbadft
 	}
 	return
 }
 
 func getFirstDate(path string, ftype int) (date time.Time, err error) {
 	f, err := os.Open(path)
 	if err != nil {
 		return
 	}
 	defer f.Close()
 	sc, err := getScanner(f, ftype)
 	if err != nil {
 		return
 	}
 	sc.Scan()
 	b := sc.Bytes()
 	if len(b) == 0 {
 		log.Printf("0 number of bytes scanned from :%s", path)
 		err = errbadscan
 	} else {
 		date, err = getDate(b, ftype)
 	}
 	return
 }
 
 func (fsa *fsarchive) visit(path string, f os.FileInfo, err error) error {
 	var (
 		ft int
 	)
 	fname := f.Name()
 	log.Print("examining: ", fname)
 	if strings.LastIndex(path, fsa.descriminator) == -1 {
 		log.Printf("visit: descriminator:%s not found in path:%s . ignoring\n", fsa.descriminator, path)
 		return nil
 	}
 	if f.Mode().IsRegular() {
 		numind := strings.IndexFunc(fname, unicode.IsDigit)
 		extind := strings.LastIndex(fname, ".bz2")
 		xmlind := strings.LastIndex(fname, ".xml")
 		if (extind == -1 && xmlind == -1) || numind == -1 {
 			log.Printf("file: %s not in foo.YYYYMMDD.HHMM.[xml].bz2... format. extind:%d numberind:%d xmlind:%d", fname, extind, numind, xmlind)
 			return nil
 		}
 		if extind-numind != 13 && xmlind-numind != 13 {
 			log.Printf("file: %s not in foo.YYYYMMDD.HHMM.[xml].bz2... format. extind:%d numberind:%d xmlind:%d", fname, extind, numind, xmlind)
 			return nil
 		}
 		if xmlind != -1 {
 			ft = xmlfile
 		} else {
 			ft = mrtfile
 		}
 		time, err := getFirstDate(path, ft)
 		if err != nil {
 			log.Print("visit: getFirstDate failed on file: ", fname, " with error: ", err)
 			//return err // this should make the filepath.Walk to end
 			return nil
 		}
 		fsa.tempentryfiles = append(fsa.tempentryfiles, archentryfile{path: path, sdate: time, sz: f.Size(), ftype: ft})
 	}
 	return nil
 }
 
 func (fsa *fsarchive) SetDelta(a time.Duration) {
 	fsa.timedelta = a
 }
 
 func NewFsArchive(path, descr string) *fsarchive {
 	return &fsarchive{
 		rootpathstr:    path,
 		entryfiles:     &timeentryslice{},
 		tempentryfiles: timeentryslice{},
 		curyr:          0,
 		curmon:         0,
 		curday:         0,
 		reqchan:        make(chan string),
 		scanning:       false,
 		Scanwg:         &sync.WaitGroup{},
 		scanch:         make(chan struct{}),
 		timedelta:      15 * time.Minute,
 		descriminator:  descr,
 		Conf:           &fsarconf{},
 	}
 }
 
 //trying to see if a dir name is in YYYY.MM form
 //returns true, year, month if it is, or false, 0, 0 if not.
 func isYearMonthDir(fname string) (res bool, yr int, mon int) {
 	var err error
 	res = false
 	yr = 0
 	mon = 0
 	isdot := func(r rune) bool {
 		if r == '.' {
 			return true
 		}
 		return false
 	}
 	ind := strings.IndexFunc(fname, isdot)
 	//not found or in the form foo.
 	if ind == -1 || ind == len(fname) {
 		return
 	}
 	//not YYYY or MM
 	if len(fname[:ind]) != 4 || len(fname[ind+1:]) != 2 {
 		return
 	}
 	yr, err = strconv.Atoi(fname[:ind])
 	if err != nil {
 		return
 	}
 	mon, err = strconv.Atoi(fname[ind+1:])
 	if err != nil {
 		return
 	}
 	if mon < 1 || mon > 12 {
 		return
 	}
 	//the values were found to be valid
 	res = true
 	return
 }
 
 func (fsa *fsarchive) printEntries() {
 	log.Printf("dumping entries")
 	for _, ef := range *fsa.entryfiles {
 		fmt.Printf("[path:%s\tdate:%s\tsize:%d\ttype:%s\n", ef.path, ef.sdate, ef.sz, ftypestrings[ef.ftype])
 	}
 }
 
 func (fsa *fsarchive) scan(ar archive) {
 	//clear the temp slice
 	fsa.tempentryfiles = []archentryfile{}
 	fsa.Scanwg.Add(1)
 	fsa.scanning = true
 	filepath.Walk(fsa.rootpathstr, ar.visit)
 	sort.Sort(fsa.tempentryfiles)
 	//allow the serve goroutine to unblock in case of STOP.
 	fsa.Scanwg.Done()
 	//signal the serve goroutine on scandone channel
 	fsa.scanch <- struct{}{}
 }
 
 func (fsa *fsarchive) Serve(wg *sync.WaitGroup, ar archive) (reqchan chan<- string) {
 	if fsa.reqchan == nil { // we have closed the channel and now called again
 		fsa.reqchan = make(chan string)
 	}
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		for {
 			select {
 			case req := <-fsa.reqchan:
 				switch req {
 				case "SCAN":
 					if fsa.scanning {
 						log.Print("fsarchive: already scanning. ignoring command")
 					} else { //fire an async goroutine to scan the files and wait for SCANDONE
 						go fsa.scan(ar)
 					}
 				case "DUMPENTRIES":
 					if fsa.scanning {
 						log.Print("fsar: warning. scanning in progress")
 					}
 					fsa.printEntries()
 				case "STOP":
 					log.Print("fsar: stopping")
 					fsa.Scanwg.Wait()
 					fsa.reqchan = nil //no more stuff from this channel
 					return
 				default:
 					log.Print("fsarchive: unknown request: ", req)
 				}
 			case <-fsa.scanch:
 				//update the reference to our file slice
 				fsa.entryfiles = &fsa.tempentryfiles
 				fsa.scanning = false
 				//let the config know
 				log.Printf("setting archive conf  from :%v to a slice of len: %v\n", fsa.Conf.ar, len(*fsa.entryfiles))
 				fsa.Conf.ar = fsa
 				log.Print("fsarchive: scan finished")
 			}
 		}
 	}()
 	return fsa.reqchan
 }