func (dp *NDT7ResultParser) Flush() error {
	return dp.Base.Flush()
}
NB: These functions are also required to complete the etl.Parser interface.
For NDT7Result, we just forward the calls to the Inserter.
package parser
import (
	"encoding/json"
	"log"
	"strings"
	"time"
	"cloud.google.com/go/bigquery"
	"cloud.google.com/go/civil"
	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/row"
	"github.com/m-lab/etl/schema"
	"github.com/m-lab/go/logx"
	"github.com/m-lab/ndt-server/ndt7/model"
)
type NDT7ResultParser struct {
	*row.Base
	table  string
	suffix string
}
func NewNDT7ResultParser(sink row.Sink, table, suffix string, ann v2as.Annotator) etl.Parser {
	bufSize := etl.NDT7.BQBufferSize()
	if ann == nil {
		ann = v2as.GetAnnotator(etl.BatchAnnotatorURL)
	}
	return &NDT7ResultParser{
		Base:   row.NewBase(table, sink, bufSize, ann),
		table:  table,
		suffix: suffix,
	}
}
func (dp *NDT7ResultParser) TaskError() error {
	stats := dp.GetStats()
	if stats.Total() < 10*stats.Failed {
		log.Printf("Warning: high row commit errors (more than 10%%): %d failed of %d accepted\n",
			stats.Failed, stats.Total())
		return etl.ErrHighInsertionFailureRate
	}
	return nil
}
func (dp *NDT7ResultParser) IsParsable(testName string, data []byte) (string, bool) {
	if strings.Contains(testName, "ndt7") && (strings.HasSuffix(testName, "json.gz") || strings.HasSuffix(testName, "json")) {
		return "ndt7_result", true
	}
	logx.Debug.Println("ndt7 unknown file:", testName)
	return "unknown", false
}
func (dp *NDT7ResultParser) ParseAndInsert(meta map[string]bigquery.Value, testName string, test []byte) error {
	metrics.WorkerState.WithLabelValues(dp.TableName(), "ndt7_result").Inc()
	defer metrics.WorkerState.WithLabelValues(dp.TableName(), "ndt7_result").Dec()
	row := schema.NDT7ResultRow{
		Parser: schema.ParseInfo{
			Version:    Version(),
			Time:       time.Now(),
			ArchiveURL: meta["filename"].(string),
			Filename:   testName,
			GitCommit:  GitCommit(),
		},
	}
	err := json.Unmarshal(test, &row.Raw)
	if err != nil {
		log.Println(meta["filename"].(string), testName, err)
		metrics.TestCount.WithLabelValues(dp.TableName(), "ndt7_result", "Unmarshal").Inc()
		return err
	}
	if row.Raw.Download != nil && row.Raw.Download.ServerMeasurements != nil {
		for i := range row.Raw.Download.ServerMeasurements {
			row.Raw.Download.ServerMeasurements[i].ConnectionInfo = nil
		}
	}
	if row.Raw.Upload != nil && row.Raw.Upload.ServerMeasurements != nil {
		for i := range row.Raw.Upload.ServerMeasurements {
			row.Raw.Upload.ServerMeasurements[i].ConnectionInfo = nil
		}
	}
	row.Date = meta["date"].(civil.Date)
	if row.Raw.Download != nil {
		row.A = downSummary(row.Raw.Download)
	} else if row.Raw.Upload != nil {
		row.A = upSummary(row.Raw.Upload)
	} else {
		metrics.WarningCount.WithLabelValues(
			dp.TableName(), "ndt7", "download and upload are both nil").Inc()
	}
	row.ID = row.A.UUID
	metrics.RowSizeHistogram.WithLabelValues(
		dp.TableName()).Observe(float64(len(test)))
	err = dp.Base.Put(&row)
	if err != nil {
		return err
	}
	metrics.TestCount.WithLabelValues(dp.TableName(), "ndt7_result", "ok").Inc()
	return nil
}
func downSummary(down *model.ArchivalData) schema.NDT7Summary {
	return schema.NDT7Summary{
		UUID:               down.UUID,
		TestTime:           down.StartTime,
		CongestionControl:  "bbr",
		MeanThroughputMbps: downRate(down.ServerMeasurements),
		MinRTT:             minRTT(down.ServerMeasurements),
		LossRate:           lossRate(down.ServerMeasurements),
	}
}
func upSummary(up *model.ArchivalData) schema.NDT7Summary {
	return schema.NDT7Summary{
		UUID:               up.UUID,
		TestTime:           up.StartTime,
		CongestionControl:  "bbr",  
		MeanThroughputMbps: upRate(up.ServerMeasurements),
		MinRTT:             minRTT(up.ServerMeasurements),
		LossRate:           0,  
	}
}
func lossRate(m []model.Measurement) float64 {
	var loss float64
	if len(m) > 0 {
		loss = float64(m[len(m)-1].TCPInfo.BytesRetrans) / float64(m[len(m)-1].TCPInfo.BytesSent)
	}
	return loss
}
func downRate(m []model.Measurement) float64 {
	var mbps float64
	if len(m) > 0 {
		mbps = 8 * float64(m[len(m)-1].TCPInfo.BytesAcked) / float64(m[len(m)-1].TCPInfo.ElapsedTime)
	}
	return mbps
}
func upRate(m []model.Measurement) float64 {
	var mbps float64
	if len(m) > 0 {
		mbps = 8 * float64(m[len(m)-1].TCPInfo.BytesReceived) / float64(m[len(m)-1].TCPInfo.ElapsedTime)
	}
	return mbps
}
func minRTT(m []model.Measurement) float64 {
	var rtt float64
	if len(m) > 0 {
		rtt = float64(m[len(m)-1].TCPInfo.MinRTT) / 1000
	}
	return rtt
}