7190

Committed 23 Feb 2022 12:10AM UTC coverage: 64.255%. First build

Build # 7190

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

Rename etl_test_count and update it for unsuccessful scamper parsing (#1053)

* Filter out legacy errors from scamper1 parsing errors

* Return ErrIsInvalid

* Use custom error

* Fix failed stats

* Revert "Fix failed stats"

This reverts commit 062f3499e.

* Use TestCount metric

* Change to etl_test_total

* Rename in worker_test.go

* Rename in worker_test.go fix

* Remove error filtering

* Add TestTotal to error

Run Details

48 of 48 new or added lines in 15 files covered. (100.0%)

3854 of 5998 relevant lines covered (64.25%)

0.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

74.67

/parser/hopannotation1.go

package parser

import (
        "encoding/json"
        "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/traceroute-caller/hopannotation"
)

//=====================================================================================
//                       HopAnnotation1 Parser
//=====================================================================================

// HopAnnotation1Parser handles parsing for the HopAnnotation1 datatype.
type HopAnnotation1Parser struct {
        *row.Base
        table  string
        suffix string
}

// NewHopAnnotation1Parser returns a new parser for the HopAnnotation1 archives.
func NewHopAnnotation1Parser(sink row.Sink, table, suffix string, ann v2as.Annotator) etl.Parser {
        bufSize := etl.HOPANNOTATION1.BQBufferSize()
        if ann == nil {
                ann = v2as.GetAnnotator(etl.BatchAnnotatorURL)
        }

        return &HopAnnotation1Parser{
                Base:   row.NewBase(table, sink, bufSize, ann),
                table:  table,
                suffix: suffix,
        }
}

// IsParsable returns the canonical test type and whether to parse data.
func (p *HopAnnotation1Parser) IsParsable(testName string, data []byte) (string, bool) {
        if strings.HasSuffix(testName, "json") {
                return "hopannotation1", true
        }
        return "", false
}

// ParseAndInsert decodes the HopAnnotation1 data and inserts it into BQ.
func (p *HopAnnotation1Parser) ParseAndInsert(fileMetadata map[string]bigquery.Value, testName string, rawContent []byte) error {
        metrics.WorkerState.WithLabelValues(p.TableName(), "hopannotation1").Inc()
        defer metrics.WorkerState.WithLabelValues(p.TableName(), "hopannotation1").Dec()

        row := schema.HopAnnotation1Row{
                Parser: schema.ParseInfo{
                        Version:    Version(),
                        Time:       time.Now(),
                        ArchiveURL: fileMetadata["filename"].(string),
                        Filename:   testName,
                        GitCommit:  GitCommit(),
                },
        }

        raw := hopannotation.HopAnnotation1{}
        err := json.Unmarshal(rawContent, &raw)
        if err != nil {
                metrics.TestTotal.WithLabelValues(p.TableName(), "hopannotation1", "decode-location-error").Inc()
                return err
        }

        // Fill in the row.
        row.ID = raw.ID
        row.Raw = &raw
        // NOTE: Civil is not TZ adjusted. It takes the year, month, and date from
        // the given timestamp, regardless of the timestamp's timezone. Since we
        // run our systems in UTC, all timestamps will be relative to UTC and as
        // will these dates.
        row.Date = fileMetadata["date"].(civil.Date)

        // Estimate the row size based on the input JSON size.
        metrics.RowSizeHistogram.WithLabelValues(p.TableName()).Observe(float64(len(rawContent)))

        // Insert the row.
        err = p.Base.Put(&row)
        if err != nil {
                return err
        }
        // Count successful inserts.
        metrics.TestTotal.WithLabelValues(p.TableName(), "hopannotation1", "ok").Inc()

        return nil
}

// NB: These functions are also required to complete the etl.Parser interface
// For HopAnnotation1, we just forward the calls to the Inserter.

func (p *HopAnnotation1Parser) Flush() error {
        return p.Base.Flush()
}

func (p *HopAnnotation1Parser) TableName() string {
        return p.table
}

func (p *HopAnnotation1Parser) FullTableName() string {
        return p.table + p.suffix
}

// RowsInBuffer returns the count of rows currently in the buffer.
func (p *HopAnnotation1Parser) RowsInBuffer() int {
        return p.GetStats().Pending
}

// Committed returns the count of rows successfully committed to BQ.
func (p *HopAnnotation1Parser) Committed() int {
        return p.GetStats().Committed
}

// Accepted returns the count of all rows received through InsertRow(s).
func (p *HopAnnotation1Parser) Accepted() int {
        return p.GetStats().Total()
}

// Failed returns the count of all rows that could not be committed.
func (p *HopAnnotation1Parser) Failed() int {
        return p.GetStats().Failed
}

1	package parser
2
3	import (
4	"encoding/json"
5	"strings"
6	"time"
7
8	"cloud.google.com/go/bigquery"
9	"cloud.google.com/go/civil"
10	v2as "github.com/m-lab/annotation-service/api/v2"
11	"github.com/m-lab/etl/etl"
12	"github.com/m-lab/etl/metrics"
13	"github.com/m-lab/etl/row"
14	"github.com/m-lab/etl/schema"
15	"github.com/m-lab/traceroute-caller/hopannotation"
16	)
17
18	//=====================================================================================
19	// HopAnnotation1 Parser
20	//=====================================================================================
21
22	// HopAnnotation1Parser handles parsing for the HopAnnotation1 datatype.
23	type HopAnnotation1Parser struct {
24	*row.Base
25	table string
26	suffix string
27	}
28
29	// NewHopAnnotation1Parser returns a new parser for the HopAnnotation1 archives.
30	func NewHopAnnotation1Parser(sink row.Sink, table, suffix string, ann v2as.Annotator) etl.Parser {	1✔
31	bufSize := etl.HOPANNOTATION1.BQBufferSize()	1✔
32	if ann == nil {	1✔
33	ann = v2as.GetAnnotator(etl.BatchAnnotatorURL)	×
34	}	×
35
36	return &HopAnnotation1Parser{	1✔
37	Base: row.NewBase(table, sink, bufSize, ann),	1✔
38	table: table,	1✔
39	suffix: suffix,	1✔
40	}	1✔
41	}
42
43	// IsParsable returns the canonical test type and whether to parse data.
44	func (p *HopAnnotation1Parser) IsParsable(testName string, data []byte) (string, bool) {	1✔
45	if strings.HasSuffix(testName, "json") {	2✔
46	return "hopannotation1", true	1✔
47	}	1✔
48	return "", false	1✔
49	}
50
51	// ParseAndInsert decodes the HopAnnotation1 data and inserts it into BQ.
52	func (p *HopAnnotation1Parser) ParseAndInsert(fileMetadata map[string]bigquery.Value, testName string, rawContent []byte) error {	1✔
53	metrics.WorkerState.WithLabelValues(p.TableName(), "hopannotation1").Inc()	1✔
54	defer metrics.WorkerState.WithLabelValues(p.TableName(), "hopannotation1").Dec()	1✔
55		1✔
56	row := schema.HopAnnotation1Row{	1✔
57	Parser: schema.ParseInfo{	1✔
58	Version: Version(),	1✔
59	Time: time.Now(),	1✔
60	ArchiveURL: fileMetadata["filename"].(string),	1✔
61	Filename: testName,	1✔
62	GitCommit: GitCommit(),	1✔
63	},	1✔
64	}	1✔
65		1✔
66	raw := hopannotation.HopAnnotation1{}	1✔
67	err := json.Unmarshal(rawContent, &raw)	1✔
68	if err != nil {	1✔
69	metrics.TestTotal.WithLabelValues(p.TableName(), "hopannotation1", "decode-location-error").Inc()	×
70	return err	×
71	}	×
72
73	// Fill in the row.
74	row.ID = raw.ID	1✔
75	row.Raw = &raw	1✔
76	// NOTE: Civil is not TZ adjusted. It takes the year, month, and date from	1✔
77	// the given timestamp, regardless of the timestamp's timezone. Since we	1✔
78	// run our systems in UTC, all timestamps will be relative to UTC and as	1✔
79	// will these dates.	1✔
80	row.Date = fileMetadata["date"].(civil.Date)	1✔
81		1✔
82	// Estimate the row size based on the input JSON size.	1✔
83	metrics.RowSizeHistogram.WithLabelValues(p.TableName()).Observe(float64(len(rawContent)))	1✔
84		1✔
85	// Insert the row.	1✔
86	err = p.Base.Put(&row)	1✔
87	if err != nil {	1✔
88	return err	×
89	}	×
90	// Count successful inserts.
91	metrics.TestTotal.WithLabelValues(p.TableName(), "hopannotation1", "ok").Inc()	1✔
92		1✔
93	return nil	1✔
94	}
95
96	// NB: These functions are also required to complete the etl.Parser interface
97	// For HopAnnotation1, we just forward the calls to the Inserter.
98
99	func (p *HopAnnotation1Parser) Flush() error {	1✔
100	return p.Base.Flush()	1✔
101	}	1✔
102
103	func (p *HopAnnotation1Parser) TableName() string {	1✔
104	return p.table	1✔
105	}	1✔
106
107	func (p *HopAnnotation1Parser) FullTableName() string {	×
108	return p.table + p.suffix	×
109	}	×
110
111	// RowsInBuffer returns the count of rows currently in the buffer.
112	func (p *HopAnnotation1Parser) RowsInBuffer() int {	×
113	return p.GetStats().Pending	×
114	}	×
115
116	// Committed returns the count of rows successfully committed to BQ.
117	func (p *HopAnnotation1Parser) Committed() int {	×
118	return p.GetStats().Committed	×
119	}	×
120
121	// Accepted returns the count of all rows received through InsertRow(s).
122	func (p *HopAnnotation1Parser) Accepted() int {	1✔
123	return p.GetStats().Total()	1✔
124	}	1✔
125
126	// Failed returns the count of all rows that could not be committed.
127	func (p *HopAnnotation1Parser) Failed() int {	×
128	return p.GetStats().Failed	×
129	}	×

m-lab / etl / 7190

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous