func GetFilename(filename string) (string, error) {
	if strings.HasPrefix(filename, "gs://") {
		return filename, nil
	}
	decode, err := base64.StdEncoding.DecodeString(filename)
	if err != nil {
		return "", errors.New("invalid file path: " + filename)
	}
	fn := string(decode[:])
	if strings.HasPrefix(fn, "gs://") {
		return fn, nil
	}
	return "", errors.New("invalid base64 encoded file path: " + fn)
}
GetFilename converts request received from the queue into a filename.
TODO(dev) Add unit test
package etl
import (
	"encoding/base64"
	"errors"
	"fmt"
	"log"
	"net"
	"regexp"
	"strings"
)
var (
	IsBatch bool
	OmitDeltas bool
	GCloudProject string
	BigqueryProject string
	BigqueryDataset string
	BatchAnnotatorURL string
)
var (
	GitCommit = "nocommit"
	Version   = "noversion"
)
const YYYYMMDD = `\d{4}[01]\d[0123]\d`
const MlabDomain = `measurement-lab.org`
const BucketPattern = `gs: 
const ExpTypePattern = `(?:([a-z-]+)/)?([a-z0-9-]+)/`  
const DatePathPattern = `(\d{4}/[01]\d/[0123]\d)/`
const dateTime = `(\d{4}[01]\d[0123]\d)T(\d{6}(\.\d{0,6})?)Z`
const type2 = `(?:-([a-z0-9-]+))?`  
const mlabNSiteNN = `-(mlab\d)-([a-z]{3}\d[0-9t])-`
const expNNNNE = `([a-z-]+)(?:-(\d{4}))?(-e)?`
const suffix = `(\.tar|\.tar.gz|\.tgz)$`
var (
	basicTaskPattern = regexp.MustCompile(
		`(?P<preamble>.*)` + dateTime + `(?P<postamble>.*)`)
	startPattern = regexp.MustCompile(`^` + BucketPattern + ExpTypePattern + DatePathPattern + `$`)
	endPattern   = regexp.MustCompile(`^` +
		type2 +  
		mlabNSiteNN +  
		expNNNNE +  
		suffix + `$`)  
	dateTimePattern = regexp.MustCompile(dateTime)
	sitePattern     = regexp.MustCompile(type2 + mlabNSiteNN)
	justSitePattern = regexp.MustCompile(`.*` + mlabNSiteNN + `.*`)
)
type DataPath struct {
	URI  string  
	Path string  
	Bucket   string  
	ExpDir   string  
	DataType string  
	DatePath string  
	PackedDate string  
	PackedTime string  
	DataType2  string  
	Host       string  
	Site       string  
	Experiment string  
	FileNumber string  
	Embargo    string  
	Suffix     string  
}
func ValidateTestPath(uri string) (DataPath, error) {
	basic := basicTaskPattern.FindStringSubmatch(uri)
	if basic == nil {
		return DataPath{}, errors.New("Path missing date-time string")
	}
	preamble := startPattern.FindStringSubmatch(basic[1])
	if preamble == nil {
		return DataPath{}, errors.New("Invalid preamble: " + fmt.Sprint(basic))
	}
	post := endPattern.FindStringSubmatch(basic[5])
	if post == nil {
		return DataPath{}, errors.New("Invalid postamble: " + basic[5])
	}
	dp := DataPath{
		URI:        uri,
		Path:       path(uri),
		Bucket:     preamble[1],
		ExpDir:     preamble[2],
		DataType:   preamble[3],
		DatePath:   preamble[4],
		PackedDate: basic[2],
		PackedTime: basic[3],
		DataType2:  post[1],
		Host:       post[2],
		Site:       post[3],
		Experiment: post[4],
		FileNumber: post[5],
		Embargo:    post[6],
		Suffix:     post[7],
	}
	dataType := dp.GetDataType()
	if dataType == INVALID {
		log.Printf("Invalid filename: %s\n", uri)
		return DataPath{}, ErrBadDataType
	}
	return dp, nil
}
func (dp DataPath) GetDataType() DataType {
	dt, ok := dirToDataType[dp.DataType]
	if !ok {
		return INVALID
	}
	return dt
}
func (dp DataPath) TableBase() string {
	return dp.GetDataType().Table()
}
func path(uri string) string {
	parts := strings.SplitN(uri, "/", 4)
	if len(parts) != 4 || parts[0] != "gs:" || len(parts[3]) == 0 {
		return ""
	}
	return parts[3]
}
func IsBatchService() bool {
	return IsBatch
}
func GetIATACode(rawFilename string) string {
	parts := justSitePattern.FindStringSubmatch(rawFilename)
	if len(parts) != 3 {
		log.Println("Unable to extract IATA code from", rawFilename)
		return ""
	}
	if len(parts[2]) < 3 {
		return parts[2]
	}
	return parts[2][0:3]
}
func GetIntFromIPv4(p4 net.IP) uint {
	return uint(p4[0])<<24 + uint(p4[1])<<16 + uint(p4[2])<<8 + uint(p4[3])
}
func GetIntFromIPv6Upper(p6 net.IP) uint64 {
	return uint64(p6[0])<<56 + uint64(p6[1])<<48 + uint64(p6[2])<<40 + uint64(p6[3])<<32 + uint64(p6[4])<<24 + uint64(p6[5])<<16 + uint64(p6[6])<<8 + uint64(p6[7])
}
func NumberBitsDifferent(first string, second string) (int, int) {
	ip1 := net.ParseIP(first)
	ip2 := net.ParseIP(second)
	if ip1.To4() != nil && ip2.To4() != nil {
		dist := uint(GetIntFromIPv4(ip1.To4()) ^ uint(GetIntFromIPv4(ip2.To4())))
		n := 0
		for ; dist != 0; dist >>= 1 {
			n++
		}
		return n, 4
	}
	if ip1.To16() != nil && ip2.To16() != nil {
		dist := uint64(GetIntFromIPv6Upper(ip1.To16())) ^ uint64(GetIntFromIPv6Upper(ip2.To16()))
		n := 0
		for ; dist != 0; dist >>= 1 {
			n++
		}
		return n, 6
	}
	return -1, 0
}
type DataType string
func (dt DataType) BQBufferSize() int {
	if dt == NDT {
		if OmitDeltas {
			return dataTypeToBQBufferSize[NDT_OMIT_DELTAS]
		}
	}
	return dataTypeToBQBufferSize[dt]
}
const (
	ANNOTATION      = DataType("annotation")
	HOPANNOTATION1  = DataType("hopannotation1")
	NDT             = DataType("ndt")
	NDT5            = DataType("ndt5")
	NDT7            = DataType("ndt7")
	NDT_OMIT_DELTAS = DataType("ndt_nodelta")  
	SS              = DataType("sidestream")
	PCAP            = DataType("pcap")
	PT              = DataType("traceroute")
	SCAMPER1        = DataType("scamper1")
	SW              = DataType("switch")
	TCPINFO         = DataType("tcpinfo")
	INVALID         = DataType("invalid")
)
var (
	dirToDataType = map[string]DataType{
		"annotation":       ANNOTATION,
		"hopannotation1":   HOPANNOTATION1,
		"ndt":              NDT,
		"ndt5":             NDT5,
		"ndt7":             NDT7,
		"sidestream":       SS,
		"paris-traceroute": PT,
		"pcap":             PCAP,
		"scamper1":         SCAMPER1,
		"switch":           SW,
		"tcpinfo":          TCPINFO,
		"traceroute":       PT,
	}
	dataTypeToTable = map[DataType]string{
		ANNOTATION:     "annotation",
		HOPANNOTATION1: "hopannotation1",
		NDT:            "ndt",
		SS:             "sidestream",
		PCAP:           "pcap",
		PT:             "traceroute",
		SCAMPER1:       "scamper1",
		SW:             "switch",
		TCPINFO:        "tcpinfo",
		NDT5:           "ndt5",
		NDT7:           "ndt7",
		INVALID:        "invalid",
	}
	dataTypeToBQBufferSize = map[DataType]int{
		ANNOTATION:      400,  
		HOPANNOTATION1:  200,
		NDT:             10,
		NDT_OMIT_DELTAS: 50,
		TCPINFO:         5,
		SS:              500,  
		PCAP:            200,
		PT:              20,
		SCAMPER1:        200,
		SW:              100,
		NDT5:            200,
		NDT7:            200,
		INVALID:         0,
	}
	dataTypeToSkipCount = map[DataType]int{
		PCAP: 9,
	}
)
func DirToTablename(dir string) string {
	return dataTypeToTable[dirToDataType[dir]]
}
func (dt DataType) SkipCount() int {
	return dataTypeToSkipCount[dt]
}
func (dt DataType) BigqueryProject() string {
	project := BigqueryProject
	if project != "" {
		return project
	}
	return GCloudProject
}
func (dt DataType) Dataset() string {
	dataset := BigqueryDataset
	if dataset != "" {
		return dataset
	}
	if IsBatchService() {
		return "batch"
	}
	return "base_tables"
}
func (dt DataType) Table() string {
	return dataTypeToTable[dt]
}