func (ndt NDTTest) AnnotateClients(annMap map[string]*api.Annotations) error {
	spec := ndt.getConnSpec()
	ip, _ := spec.GetString([]string{"client_ip"})
	if ip == "" {
		return ErrAnnotationError
	}
	if data, ok := annMap[ip]; ok && data.Geo != nil {
		CopyStructToMap(data.Geo, spec.Get("client_geolocation"))
		if data.Network != nil {
			asn, err := data.Network.BestASN()
			if err != nil {
				metrics.AnnotationErrorCount.With(prometheus.
					Labels{"source": "NDTTest BestASN error on client IP."}).Inc()
			} else {
				spec.Get("client").Get("network")["asn"] = asn
				c := v2as.ConvertAnnotationsToClientAnnotations(data)
				copyStructToMapDirectly(c.Geo, spec.Get("ClientX").Get("Geo"))
				copyStructToMapDirectly(c.Network, spec.Get("ClientX").Get("Network"))
			}
		}
	} else {
		if ip == "45.56.98.222" {
			metrics.AnnotationErrorCount.With(prometheus.
				Labels{"source": "NDTTest missing annotations for client IP " + ip}).Inc()
			return ErrAnnotationError
		}
		logMissing.Println("Missing annotation for", ip)
		if strings.HasPrefix(ip, "2002:") {
			metrics.AnnotationErrorCount.With(prometheus.
				Labels{"source": "NDTTest missing annotations for client IP in 2002:..."}).Inc()
			return ErrAnnotationError
		}
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "NDTTest missing annotations for client IP"}).Inc()
		return ErrAnnotationError
	}
	return nil
}
AnnotateClients adds the client annotations. See parser.Annotatable
This is a bit ugly because of the use of bigquery Value maps.
package parser
import (
	"errors"
	"fmt"
	"log"
	"os"
	"reflect"
	"regexp"
	"strconv"
	"strings"
	"time"
	"github.com/m-lab/go/logx"
	"cloud.google.com/go/bigquery"
	"github.com/m-lab/annotation-service/api"
	v2as "github.com/m-lab/annotation-service/api/v2"
	"github.com/m-lab/etl/etl"
	"github.com/m-lab/etl/metrics"
	"github.com/m-lab/etl/schema"
	"github.com/m-lab/etl/web100"
	"github.com/prometheus/client_golang/prometheus"
)
var (
	NDTEstimateBW, _ = strconv.ParseBool(os.Getenv("NDT_ESTIMATE_BW"))
)
const (
	minNumSnapshots = 1600  
	maxNumSnapshots = 2800  
)
const dateDir = `^(?P<dir>\d{4}/\d{2}/\d{2}/)?`
const dateField = `(?P<date>\d{8})`
const timeField = `(?P<time>[012]\d:[0-6]\d:\d{2}\.\d{1,10})`
const address = `(?P<address>.*)`
const suffix = `(?P<suffix>[a-z2].*)`
var (
	testFilePattern = regexp.MustCompile(
		"^" + dateDir + dateField + "T" + timeField + "Z_" + address + `\.` + suffix + "$")
	gzTestFilePattern = regexp.MustCompile(
		"^" + dateDir + dateField + "T" + timeField + "Z_" + address + `\.` + suffix + `\.gz$`)
	datePattern = regexp.MustCompile(dateField)
	timePattern = regexp.MustCompile("T" + timeField + "Z_")
	endPattern  = regexp.MustCompile(suffix + `$`)
)
type TestInfo struct {
	DateDir   string     
	Date      string     
	Time      string     
	Address   string     
	Suffix    string     
	Timestamp time.Time  
}
func ParseNDTFileName(path string) (*TestInfo, error) {
	fields := gzTestFilePattern.FindStringSubmatch(path)
	if fields == nil {
		fields = testFilePattern.FindStringSubmatch(path)
	}
	if fields == nil {
		if !datePattern.MatchString(path) {
			return nil, errors.New("Path should contain yyyymmddT: " + path)
		} else if !timePattern.MatchString(path) {
			return nil, errors.New("Path should contain Thh:mm:ss.ff...Z_: " + path)
		} else if !endPattern.MatchString(path) {
			return nil, errors.New("Path should end in \\.[a-z2].*: " + path)
		}
		return nil, errors.New("Invalid test path: " + path)
	}
	timestamp, err := time.Parse("20060102T15:04:05.999999999Z_", fields[2]+"T"+fields[3]+"Z_")
	if err != nil {
		log.Println(fields[2] + "T" + fields[3] + "   " + err.Error())
		return nil, errors.New("Invalid test path: " + path)
	}
	return &TestInfo{fields[1], fields[2], fields[3], fields[4], fields[5], timestamp}, nil
}
type fileInfoAndData struct {
	fn   string
	info TestInfo
	data []byte
}
type NDTParser struct {
	Base
	taskFileName string  
	timestamp    string  
	c2s *fileInfoAndData
	s2c *fileInfoAndData
	metaFile *MetaFileData
}
func NewNDTParser(ins etl.Inserter, annotator ...v2as.Annotator) *NDTParser {
	bufSize := etl.NDT.BQBufferSize()
	var ann v2as.Annotator
	if len(annotator) > 0 && annotator[0] != nil {
		ann = annotator[0]
	} else {
		ann = v2as.GetAnnotator(etl.BatchAnnotatorURL)
	}
	return &NDTParser{Base: *NewBase(ins, bufSize, ann)}
}
func (n *NDTParser) TaskError() error {
	if n.Inserter.Committed() < 10*n.Inserter.Failed() {
		log.Printf("Warning: high row commit errors: %d / %d\n",
			n.Inserter.Accepted(), n.Inserter.Failed())
		return errors.New("too many insertion failures")
	}
	return nil
}
func (n *NDTParser) Flush() error {
	if n.timestamp != "" {
		n.processGroup()
	}
	return n.Base.AnnotateAndFlush(n.TableBase())
}
func (n *NDTParser) TableName() string {
	return n.TableBase()
}
func (n *NDTParser) IsParsable(testName string, data []byte) (string, bool) {
	info, err := ParseNDTFileName(testName)
	if err != nil {
		return "unknown", false
	}
	switch info.Suffix {
	case "c2s_snaplog":
		fallthrough  
	case "s2c_snaplog":
		fallthrough  
	case "meta":
		return info.Suffix, true
	case "c2s_ndttrace":
		fallthrough  
	case "s2c_ndttrace":
		fallthrough  
	case "cputime":
		return info.Suffix, false
	}
	metrics.TestCount.WithLabelValues(
		n.TableName(), "unknown", "unknown suffix").Inc()
	return "unknown", false
}
func (n *NDTParser) ParseAndInsert(taskInfo map[string]bigquery.Value, testName string, content []byte) error {
	info, err := ParseNDTFileName(testName)
	if err != nil {
		metrics.TestCount.WithLabelValues(
			n.TableName(), "unknown", "bad filename").Inc()
		log.Println(err)
		return nil
	}
	if info.Time != n.timestamp {
		n.processGroup()
		if info.Time < n.timestamp {
			metrics.ErrorCount.WithLabelValues(
				n.TableName(), "unknown", "TIMESTAMPS OUT OF ORDER").Inc()
			log.Printf("Timestamps out of order in: %s: %s\n",
				n.taskFileName, err)
			panic("Timestamps out of order in tar file")
		}
		n.taskFileName = taskInfo["filename"].(string)
		n.timestamp = info.Time
	} else {
		if n.taskFileName != taskInfo["filename"].(string) {
			metrics.TestCount.WithLabelValues(
				n.TableName(), "any", "inconsistent taskFileName").Inc()
		}
	}
	switch info.Suffix {
	case "c2s_snaplog":
		if n.c2s == nil {
			n.c2s = &fileInfoAndData{testName, *info, content}
		} else {
			if (n.c2s.fn + ".gz") == testName {
				n.c2s = &fileInfoAndData{testName, *info, content}
			} else if n.c2s.fn == (testName + ".gz") {
			} else {
				metrics.WarningCount.WithLabelValues(
					n.TableName(), "c2s", "timestamp collision").Inc()
				log.Printf("Collision: %s and %s\n", n.c2s.fn, testName)
			}
		}
	case "s2c_snaplog":
		if n.s2c == nil {
			n.s2c = &fileInfoAndData{testName, *info, content}
		} else {
			if (n.s2c.fn + ".gz") == testName {
				n.s2c = &fileInfoAndData{testName, *info, content}
			} else if n.s2c.fn == (testName + ".gz") {
			} else {
				metrics.WarningCount.WithLabelValues(
					n.TableName(), "s2c", "timestamp collision").Inc()
				log.Printf("Collision: %s and %s\n", n.s2c.fn, testName)
			}
		}
	case "meta":
		if n.metaFile != nil {
			metrics.WarningCount.WithLabelValues(
				n.TableName(), "meta", "timestamp collision").Inc()
		}
		n.metaFile = ProcessMetaFile(
			n.TableName(), n.TableSuffix(), testName, content)
	default:
		metrics.TestCount.WithLabelValues(
			n.TableName(), "unknown", "unparsable file").Inc()
		return errors.New("Unknown test suffix: " + info.Suffix)
	}
	return nil
}
func (n *NDTParser) reportAnomalies() {
	tag := ""
	code := 0
	if n.metaFile != nil {
		tag += "meta, "
		code += 4
	}
	if n.s2c != nil {
		tag += "s2c, "
		code += 2
	}
	if n.c2s != nil {
		tag += "c2s"
		code++
	}
	if code != 7 {
		if tag == "" {
			tag = "Found no files"
		} else {
			tag = "Found only " + tag
		}
		metrics.WarningCount.WithLabelValues(
			n.TableName(), "group", tag).Inc()
	}
}
func (n *NDTParser) processGroup() {
	n.reportAnomalies()
	if n.s2c != nil {
		n.processTest(n.s2c, "s2c")
	}
	if n.c2s != nil {
		n.processTest(n.c2s, "c2s")
	}
	n.taskFileName = ""
	n.timestamp = ""
	n.s2c = nil
	n.c2s = nil
	n.metaFile = nil
}
func (n *NDTParser) processTest(test *fileInfoAndData, testType string) {
	if len(test.data) > 10*1024*1024 {
		metrics.ErrorCount.WithLabelValues(
			n.TableName(), testType, ">10MB").Inc()
		log.Printf("Ignoring oversize snaplog: %d, %s\n",
			len(test.data), test.fn)
		return
	}
	if len(test.data) < 16*1024 {
		metrics.WarningCount.WithLabelValues(
			n.TableName(), testType, "<16KB").Inc()
		log.Printf("Note: small rawSnapLog: %d, %s\n",
			len(test.data), test.fn)
	}
	if len(test.data) == 4096 {
		metrics.WarningCount.WithLabelValues(
			n.TableName(), testType, "4KB").Inc()
	}
	metrics.WorkerState.WithLabelValues(n.TableName(), "ndt").Inc()
	defer metrics.WorkerState.WithLabelValues(n.TableName(), "ndt").Dec()
	n.getAndInsertValues(test, testType)
}
func (n *NDTParser) getDeltas(snaplog *web100.SnapLog, testType string) ([]schema.Web100ValueMap, int) {
	deltas := []schema.Web100ValueMap{}
	deltaFieldCount := 0
	if etl.OmitDeltas {
		return deltas, deltaFieldCount
	}
	snapshotCount := 0
	last := &web100.Snapshot{}
	for count := 0; count < snaplog.SnapCount() && count < maxNumSnapshots; count++ {
		snap, err := snaplog.Snapshot(count)
		if err != nil {
			metrics.TestCount.WithLabelValues(
				n.TableName(), testType, "snapshot failure").Inc()
			return nil, 0
		}
		delta := schema.EmptySnap10()
		snap.SnapshotDeltas(last, delta)
		if err != nil {
			metrics.ErrorCount.WithLabelValues(
				n.TableName(), testType, "snapValues failure").Inc()
			return nil, 0
		}
		delete(delta, "TimeStamps")
		delete(delta, "StartTimeStamp")
		delete(delta, "StartTimeUsec")
		delete(delta, "LocalAddress")
		delete(delta, "LocalAddressType")
		delete(delta, "LocalPort")
		delete(delta, "RemAddress")
		delete(delta, "RemPort")
		delete(delta, "SACK")
		if len(delta) == 1 {
			_, ok := delta["Duration"]
			if ok {
				continue
			}
		}
		delta["snapshot_num"] = count
		delta["delta_index"] = snapshotCount
		snapshotCount++
		metrics.DeltaNumFieldsHistogram.WithLabelValues(n.TableName()).
			Observe(float64(len(delta)))
		deltaFieldCount += len(delta)
		deltas = append(deltas, delta)
		last = &snap
	}
	if len(deltas) > 0 {
		deltas[len(deltas)-1]["is_last"] = true
	}
	return deltas, deltaFieldCount
}
func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
	metrics.WorkerState.WithLabelValues(n.TableName(), "ndt-parse").Inc()
	defer metrics.WorkerState.WithLabelValues(n.TableName(), "ndt-parse").Dec()
	if !strings.HasSuffix(test.fn, ".gz") {
		metrics.WarningCount.WithLabelValues(
			n.TableName(), testType, "uncompressed file").Inc()
	}
	snaplog, err := web100.NewSnapLog(test.data)
	if err != nil {
		metrics.ErrorCount.WithLabelValues(
			n.TableName(), testType, "snaplog failure").Inc()
		log.Printf("Unable to parse snaplog for %s, when processing: %s\n%s\n",
			test.fn, n.taskFileName, err)
		return
	}
	valid := true
	err = snaplog.ValidateSnapshots()
	if err != nil {
		log.Printf("ValidateSnapshots failed for %s, when processing: %s (%s)\n",
			test.fn, n.taskFileName, err)
		metrics.WarningCount.WithLabelValues(
			n.TableName(), testType, "validate failed").Inc()
		valid = false
	}
	var deltas []schema.Web100ValueMap
	deltaFieldCount := 0
	deltas, deltaFieldCount = n.getDeltas(snaplog, testType)
	if deltas == nil {
		return
	}
	final := snaplog.SnapCount() - 1
	if final > maxNumSnapshots {
		final = maxNumSnapshots
	}
	snap, err := snaplog.Snapshot(final)
	if err != nil {
		metrics.ErrorCount.WithLabelValues(
			n.TableName(), testType, "final snapshot failure").Inc()
		metrics.TestCount.WithLabelValues(
			n.TableName(), testType, "final snapshot failure").Inc()
		return
	}
	snapValues := schema.EmptySnap()
	snap.SnapshotValues(snapValues)
	if err != nil {
		metrics.ErrorCount.WithLabelValues(
			n.TableName(), testType, "final snapValues failure").Inc()
		metrics.TestCount.WithLabelValues(
			n.TableName(), testType, "final snapValues failure").Inc()
		log.Printf("Error calling SnapshotValues() in test %s, when processing: %s\n%s\n",
			test.fn, n.taskFileName, err)
		return
	}
	nestedConnSpec := make(schema.Web100ValueMap, 6)
	snaplog.ConnectionSpecValues(nestedConnSpec)
	results := schema.NewWeb100MinimalRecord(
		snaplog.Version, int64(snaplog.LogTime),
		nestedConnSpec, snapValues, deltas)
	results["id"] = ndtWeb100SyntheticUUID(test.fn)
	results["test_id"] = test.fn
	results["task_filename"] = n.taskFileName
	if snaplog.SnapCount() > maxNumSnapshots || snaplog.SnapCount() < minNumSnapshots {
		results["anomalies"].(schema.Web100ValueMap)["num_snaps"] = snaplog.SnapCount()
	}
	if !valid {
		results["anomalies"].(schema.Web100ValueMap)["snaplog_error"] = true
	}
	if NDTEstimateBW {
		congEvents := make(schema.Web100ValueMap, 10)
		snapNums, snapErr := snaplog.ChangeIndices("SmoothedRTT")
		if snapErr != nil {
			log.Println(snapErr)
		} else {
			congEvents["indices"] = snapNums
			congEvents["smoothedRTT"] = snaplog.SliceIntField("SmoothedRTT", snapNums)
			congEvents["thruOctetsAcked"] = snaplog.SliceIntField("HCThruOctetsAcked", snapNums)
			results["slices"] = congEvents
		}
	}
	lt, err := test.info.Timestamp.MarshalText()
	if err != nil {
		log.Println(err)
		metrics.ErrorCount.WithLabelValues(
			n.TableBase(), "log_time marshal error").Inc()
	} else {
		results["log_time"] = string(lt)
	}
	now, err := time.Now().MarshalText()
	if err != nil {
		log.Println(err)
		metrics.ErrorCount.WithLabelValues(
			n.TableBase(), "parse_time marshal error").Inc()
	} else {
		results["parse_time"] = string(now)
	}
	results["parser_version"] = Version()
	connSpec := schema.EmptyConnectionSpec()
	if n.metaFile != nil {
		n.metaFile.PopulateConnSpec(connSpec)
	} else {
		metrics.WarningCount.WithLabelValues(
			n.TableName(), testType, "no meta").Inc()
		results["anomalies"].(schema.Web100ValueMap)["no_meta"] = true
	}
	switch testType {
	case "c2s":
		connSpec.SetInt64("data_direction", CLIENT_TO_SERVER)
	case "s2c":
		connSpec.SetInt64("data_direction", SERVER_TO_CLIENT)
	default:
	}
	results["connection_spec"] = connSpec
	n.fixValues(results)
	metrics.EntryFieldCountHistogram.WithLabelValues(n.TableName()).
		Observe(float64(deltaFieldCount))
	if deltaFieldCount > 43000 {
		log.Printf("Lots of fields (%d) processing %s from %s\n",
			deltaFieldCount, test.fn, n.taskFileName)
	}
	dp, _ := etl.ValidateTestPath(results["task_filename"].(string))
	connSpec.Get("ServerX")["Site"] = dp.Site
	connSpec.Get("ServerX")["Machine"] = dp.Host
	ndtTest := NDTTest{results}
	err = n.Base.AddRow(ndtTest)
	if err == etl.ErrBufferFull {
		n.Base.AnnotateAndPutAsync(n.TableBase())
		err = n.Base.AddRow(ndtTest)
	}
	if err != nil {
		metrics.ErrorCount.WithLabelValues(
			n.TableName(), testType, "insert-err").Inc()
		log.Println("insert-err: " + err.Error())
		return
	}
	metrics.TestCount.WithLabelValues(
		n.TableName(), testType, "ok").Inc()
}
const (
	WC_ADDRTYPE_IPV4 = 1
	WC_ADDRTYPE_IPV6 = 2
	LOCAL_AF_IPV4    = 0
	LOCAL_AF_IPV6    = 1
)
func (n *NDTParser) fixValues(r schema.Web100ValueMap) {
	connSpec := r.GetMap([]string{"connection_spec"})
	logEntry := r.GetMap([]string{"web100_log_entry"})
	snap := logEntry.GetMap([]string{"snap"})
	nestedConnSpec := logEntry.GetMap([]string{"connection_spec"})
	if connSpec["client_hostname"] == "No FQDN name" {
		delete(connSpec, "client_hostname")
	}
	data, err := etl.ValidateTestPath(n.taskFileName)
	if err != nil {
		log.Printf("WARNING: taskFileName is unexpectedly invalid: %s %s: %q",
			n.taskFileName, n.timestamp, err)
	} else {
		connSpec.Get("server").SetString("iata_code", strings.ToUpper(data.Site[0:3]))
		hn, ok := connSpec["server_hostname"]
		if !ok || hn == "" {
			connSpec.SetString("server_hostname", fmt.Sprintf(
				"%s.%s.%s", data.Host, data.Site, etl.MlabDomain))
		}
	}
	logEntry.SubstituteString(true, []string{"connection_spec", "local_ip"},
		[]string{"snap", "LocalAddress"})
	logEntry.SubstituteString(true, []string{"connection_spec", "remote_ip"},
		[]string{"snap", "RemAddress"})
	localAddrType, ok := snap["LocalAddressType"]
	if ok {
		switch localAddrType {
		case WC_ADDRTYPE_IPV4:
			nestedConnSpec.SetInt64("local_af", LOCAL_AF_IPV4)
		case WC_ADDRTYPE_IPV6:
			nestedConnSpec.SetInt64("local_af", LOCAL_AF_IPV6)
		default:
		}
	}
	r.SubstituteString(false, []string{"connection_spec", "server_ip"},
		[]string{"web100_log_entry", "connection_spec", "local_ip"})
	r.SubstituteInt64(false, []string{"connection_spec", "server_af"},
		[]string{"web100_log_entry", "connection_spec", "local_af"})
	r.SubstituteString(false, []string{"connection_spec", "client_ip"},
		[]string{"web100_log_entry", "connection_spec", "remote_ip"})
	r.SubstituteInt64(false, []string{"connection_spec", "client_af"},
		[]string{"web100_log_entry", "connection_spec", "local_af"})
	start, ok := snap.GetInt64([]string{"StartTimeStamp"})
	if ok {
		start = 1000000 * start
		usec, ok := snap.GetInt64([]string{"StartTimeUsec"})
		if ok {
			start += usec
		}
		snap.SetInt64("StartTimeStamp", start)
	}
}
type NDTTest struct {
	schema.Web100ValueMap
}
func (ndt NDTTest) getConnSpec() schema.Web100ValueMap {
	return ndt.GetMap([]string{"connection_spec"})
}
func (ndt NDTTest) GetLogTime() time.Time {
	var t time.Time
	tt, ok := ndt.GetString([]string{"log_time"})
	if !ok {
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "NDTTest missing log_time."}).Inc()
		return time.Now()
	}
	err := t.UnmarshalText([]byte(tt))
	if err != nil {
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "NDTTest error parsing log_time."}).Inc()
		return time.Now()
	}
	return t
}
func (ndt NDTTest) GetClientIPs() []string {
	connSpec := ndt.getConnSpec()
	ip, _ := connSpec.GetString([]string{"client_ip"})
	if ip == "" {
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "NDTTest missing client IP."}).Inc()
		return []string{}
	}
	return []string{ip}
}
func (ndt NDTTest) GetServerIP() string {
	connSpec := ndt.getConnSpec()
	ip, ok := connSpec.GetString([]string{"server_ip"})
	if !ok {
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "NDTTest missing server IP."}).Inc()
		return ""
	}
	return ip
}
func CopyStructToMap(sourceStruct interface{}, destinationMap map[string]bigquery.Value) {
	structToCopy := reflect.ValueOf(sourceStruct).Elem()
	typeOfStruct := structToCopy.Type()
	for i := 0; i < typeOfStruct.NumField(); i++ {
		f := structToCopy.Field(i)
		v := f.Interface()
		switch t := v.(type) {
		case string:
			if t == "" {
				continue
			}
		case int64:
			if t == 0 {
				continue
			}
		}
		jsonTag, ok := typeOfStruct.Field(i).Tag.Lookup("json")
		name := strings.ToLower(typeOfStruct.Field(i).Name)
		if ok {
			tags := strings.Split(jsonTag, ",")
			if len(tags) > 0 && tags[0] != "" {
				name = tags[0]
			}
		}
		destinationMap[strings.ToLower(name)] = v
	}
}
func copyStructToMapDirectly(sourceStruct interface{}, destinationMap map[string]bigquery.Value) {
	structToCopy := reflect.ValueOf(sourceStruct).Elem()
	typeOfStruct := structToCopy.Type()
	for i := 0; i < typeOfStruct.NumField(); i++ {
		f := structToCopy.Field(i)
		v := f.Interface()
		switch t := v.(type) {
		case string:
			if t == "" {
				continue
			}
		case int64:
			if t == 0 {
				continue
			}
		case uint32:
			if t == 0 {
				continue
			}
			v = int64(t)
		}
		destinationMap[typeOfStruct.Field(i).Name] = v
	}
}
var logMissing = logx.NewLogEvery(nil, 60*time.Second)  