7189

Committed 23 Feb 2022 12:10AM UTC coverage: 64.221% (+0.04%) from 64.182%

Build # 7189

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/m-lab/etl/commit/<a class=hub.com/m-lab/etl/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/etl/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/etl/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/etl/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/etl/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/etl/commit/56a26697652847b66197a184d778f22573b85685">56a266976">&lt;a href=&quot;https://github.com/m-lab/etl/commit/</a><a class="double-link" href="https://github.com/m-lab/etl/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/m-lab/etl/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/m-lab/etl/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/m-lab/etl/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/m-lab/etl/commit/56a26697652847b66197a184d778f22573b85685">56a266976</a><a href="https://github.com/m-lab/etl/commit/56a26697652847b66197a184d778f22573b85685">&lt;a href=&quot;https://github.com/m-lab/etl/commit/56a26697652847b66197a184d778f22573b85685&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/m-lab/etl/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/m-lab/etl/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/m-lab/etl/commit/56a26697652847b66197a184d778f22573b85685&amp;quot;&amp;gt;56a266976&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/m-lab/etl/commit/56a26697652847b66197a184d778f22573b85685&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/m-lab/etl/commit/56a26697652847b66197a184d778f22573b85685&amp;amp;quot;&amp;amp;gt;&amp;amp;amp;quot;&amp;amp;amp;gt;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 &amp;amp;amp;amp;quot;Fix failed </a>stats&lt;/a&gt;&amp;amp;quot;

This reverts commit &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&am... (continued)

Run Details

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

3852 of 5998 relevant lines covered (64.22%)

0.69 hits per line

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

66.52

/row/row.go

package row

// TODO integrate this functionality into the go code.
// Probably should have Base implement Parser.

import (
        "context"
        "errors"
        "io"
        "log"
        "sync"
        "time"

        "github.com/prometheus/client_golang/prometheus"

        "github.com/m-lab/annotation-service/api"
        v2as "github.com/m-lab/annotation-service/api/v2"
        "github.com/m-lab/go/logx"

        "github.com/m-lab/etl/metrics"
)

// Errors that may be returned by Buffer functions.
var (
        ErrAnnotationError = errors.New("Annotation error")
        ErrNotAnnotatable  = errors.New("object does not implement Annotatable")
        ErrBufferFull      = errors.New("Buffer full")
        ErrInvalidSink     = errors.New("Not a valid row.Sink")
)

// Annotatable interface enables integration of annotation into parser.Base.
// The row type should implement the interface, and the annotations will be added
// prior to insertion.
type Annotatable interface {
        GetLogTime() time.Time
        GetClientIPs() []string // This is a slice to support mutliple hops in traceroute data.
        GetServerIP() string
        AnnotateClients(map[string]*api.Annotations) error // Must properly handle missing annotations.
        AnnotateServer(*api.Annotations) error             // Must properly handle nil parameter.
}

// Stats contains stats about buffer history.
type Stats struct {
        Buffered  int // rows buffered but not yet sent.
        Pending   int // pending counts previously buffered rows that are being committed.
        Committed int
        Failed    int
}

// Total returns the total number of rows handled.
func (s Stats) Total() int {
        return s.Buffered + s.Pending + s.Committed + s.Failed
}

// ActiveStats is a stats object that supports updates.
type ActiveStats struct {
        lock sync.RWMutex // Protects all Stats fields.
        Stats
}

// GetStats implements HasStats()
func (as *ActiveStats) GetStats() Stats {
        as.lock.RLock()
        defer as.lock.RUnlock()
        return as.Stats
}

// MoveToPending increments the Pending field.
func (as *ActiveStats) MoveToPending(n int) {
        as.lock.Lock()
        defer as.lock.Unlock()
        as.Buffered -= n
        if as.Buffered < 0 {
                log.Println("BROKEN - negative buffered")
        }
        as.Pending += n
}

// Inc increments the Buffered field
func (as *ActiveStats) Inc() {
        logx.Debug.Println("IncPending")
        as.lock.Lock()
        defer as.lock.Unlock()
        as.Buffered++
}

// Done updates the pending to failed or committed.
func (as *ActiveStats) Done(n int, err error) {
        as.lock.Lock()
        defer as.lock.Unlock()
        as.Pending -= n
        if as.Pending < 0 {
                log.Println("BROKEN: negative Pending")
        }
        if err != nil {
                as.Failed += n
        } else {
                as.Committed += n
        }
        logx.Debug.Printf("Done %d->%d %v\n", as.Pending+n, as.Pending, err)
}

// HasStats can provide stats
type HasStats interface {
        GetStats() Stats
}

// Sink defines the interface for committing rows.
// Returns the number of rows successfully committed, and error.
// Implementations should be threadsafe.
type Sink interface {
        Commit(rows []interface{}, label string) (int, error)
        io.Closer
}

// Buffer provides all basic functionality generally needed for buffering, annotating, and inserting
// rows that implement Annotatable.
// Buffer functions are THREAD-SAFE
type Buffer struct {
        lock sync.Mutex
        size int // Number of rows before starting new buffer.
        rows []interface{}
}

// NewBuffer returns a new buffer of the desired size.
func NewBuffer(size int) *Buffer {
        return &Buffer{size: size, rows: make([]interface{}, 0, size)}
}

// Append appends a row to the buffer.
// If buffer is full, this returns the buffered rows, and saves provided row
// in new buffer.  Client MUST handle the returned rows.
func (buf *Buffer) Append(row interface{}) []interface{} {
        buf.lock.Lock()
        defer buf.lock.Unlock()
        if len(buf.rows) < buf.size {
                buf.rows = append(buf.rows, row)
                return nil
        }
        rows := buf.rows
        buf.rows = make([]interface{}, 0, buf.size)
        buf.rows = append(buf.rows, row)

        return rows
}

// Reset clears the buffer, returning all pending rows.
func (buf *Buffer) Reset() []interface{} {
        buf.lock.Lock()
        defer buf.lock.Unlock()
        res := buf.rows
        buf.rows = make([]interface{}, 0, buf.size)
        return res
}

type annotator struct {
        v2 v2as.Annotator
}

// label is used to label metrics and errors in GetAnnotations
func (ann *annotator) annotateServers(rows []interface{}, label string) error {
        serverIPs := make(map[string]struct{})
        logTime := time.Time{}
        for i := range rows {
                r, ok := rows[i].(Annotatable)
                if !ok {
                        return ErrNotAnnotatable
                }

                // Only ask for the IP if it is non-empty.
                ip := r.GetServerIP()
                if ip != "" {
                        serverIPs[ip] = struct{}{}
                }

                if (logTime == time.Time{}) {
                        logTime = r.GetLogTime()
                }
        }

        ipSlice := make([]string, 0, len(rows))
        for ip := range serverIPs {
                ipSlice = append(ipSlice, ip)
        }
        if len(ipSlice) == 0 {
                return nil
        }
        response, err := ann.v2.GetAnnotations(context.Background(), logTime, ipSlice, label)
        if err != nil {
                log.Println("error in server GetAnnotations: ", err)
                metrics.AnnotationErrorCount.With(prometheus.
                        Labels{"source": "Server IP: RPC err in GetAnnotations."}).Inc()
                return err
        }
        annMap := response.Annotations
        if annMap == nil {
                log.Println("empty server annotation response")
                metrics.AnnotationErrorCount.With(prometheus.
                        Labels{"source": "Server IP: empty response"}).Inc()
                return ErrAnnotationError
        }

        for i := range rows {
                r, ok := rows[i].(Annotatable)
                if !ok {
                        err = ErrNotAnnotatable
                } else {
                        ip := r.GetServerIP()
                        if ip != "" {
                                ann, ok := annMap[ip]
                                if ok {
                                        r.AnnotateServer(ann)
                                }
                        }
                }
        }

        return err
}

var logEmptyAnn = logx.NewLogEvery(nil, 60*time.Second)

// label is used to label metrics and errors in GetAnnotations
func (ann *annotator) annotateClients(rows []interface{}, label string) error {
        ipSlice := make([]string, 0, 2*len(rows)) // This may be inadequate, but its a reasonable start.
        logTime := time.Time{}
        for i := range rows {
                r, ok := rows[i].(Annotatable)
                if !ok {
                        return ErrNotAnnotatable
                }
                ipSlice = append(ipSlice, r.GetClientIPs()...)
                if (logTime == time.Time{}) {
                        logTime = r.GetLogTime()
                }
        }

        response, err := ann.v2.GetAnnotations(context.Background(), logTime, ipSlice, label)
        if err != nil {
                log.Println("error in client GetAnnotations: ", err)
                metrics.AnnotationErrorCount.With(prometheus.
                        Labels{"source": "Client IP: RPC err in GetAnnotations."}).Inc()
                return err
        }
        annMap := response.Annotations
        if annMap == nil {
                logEmptyAnn.Println("empty client annotation response")
                metrics.AnnotationErrorCount.With(prometheus.
                        Labels{"source": "Client IP: empty response"}).Inc()
                return ErrAnnotationError
        }

        for i := range rows {
                r, ok := rows[i].(Annotatable)
                if !ok {
                        err = ErrNotAnnotatable
                } else {
                        // Will not error because we check for nil annMap above.
                        r.AnnotateClients(annMap)
                }
        }

        return err
}

// Annotate fetches and applies annotations for all rows
func (ann *annotator) Annotate(rows []interface{}, metricLabel string) error {
        metrics.WorkerState.WithLabelValues(metricLabel, "annotate").Inc()
        defer metrics.WorkerState.WithLabelValues(metricLabel, "annotate").Dec()
        if len(rows) == 0 {
                return nil
        }
        // TODO replace this with a histogram.
        defer func(label string, start time.Time) {
                metrics.AnnotationTimeSummary.With(prometheus.Labels{"test_type": label}).Observe(float64(time.Since(start).Nanoseconds()))
        }(metricLabel, time.Now())

        // TODO Consider doing these in parallel?
        clientErr := ann.annotateClients(rows, metricLabel)
        serverErr := ann.annotateServers(rows, metricLabel)

        if clientErr != nil {
                return clientErr
        }

        if serverErr != nil {
                return serverErr
        }

        return nil
}

// Base provides common parser functionality.
// Base is NOT THREAD-SAFE
type Base struct {
        sink  Sink
        ann   annotator
        buf   *Buffer
        label string // Used in metrics and errors.

        stats ActiveStats
}

// NewBase creates a new Base.  This will generally be embedded in a type specific parser.
func NewBase(label string, sink Sink, bufSize int, ann v2as.Annotator) *Base {
        buf := NewBuffer(bufSize)
        return &Base{sink: sink, ann: annotator{ann}, buf: buf, label: label}
}

// GetStats returns the buffer/sink stats.
func (pb *Base) GetStats() Stats {
        return pb.stats.GetStats()
}

// TaskError return the task level error, based on failed rows, or any other criteria.
func (pb *Base) TaskError() error {
        return nil
}

var logAnnError = logx.NewLogEvery(nil, 60*time.Second)

func (pb *Base) commit(rows []interface{}) error {
        err := pb.ann.Annotate(rows, pb.label)
        if err != nil {
                logAnnError.Println("annotation: ", err)
        }

        // TODO do we need these to be done in order.
        // This is synchronous, blocking, and thread safe.
        done, err := pb.sink.Commit(rows, pb.label)
        if done > 0 {
                pb.stats.Done(done, nil)
        }
        if err != nil {
                log.Println(pb.label, err)
                pb.stats.Done(len(rows)-done, err)
        }
        return err
}

// Flush synchronously flushes any pending rows.
func (pb *Base) Flush() error {
        rows := pb.buf.Reset()
        pb.stats.MoveToPending(len(rows))
        return pb.commit(rows)
}

// Put adds a row to the buffer.
// Iff the buffer is already full the prior buffered rows are
// annotated and committed to the Sink.
// NOTE: There is no guarantee about ordering of writes resulting from
// sequential calls to Put.  However, once a block of rows is submitted
// to pb.commit, it should be written in the same order to the Sink.
// TODO improve Annotatable architecture.
func (pb *Base) Put(row Annotatable) error {
        rows := pb.buf.Append(row)
        pb.stats.Inc()

        if rows != nil {
                pb.stats.MoveToPending(len(rows))
                err := pb.commit(rows)
                if err != nil {
                        // Note that error is likely associated with buffered rows, not the current
                        // row.
                        // When using GCS output, this may result in a corrupted json file.
                        // In that event, the test count may become meaningless.
                        metrics.TestTotal.WithLabelValues(pb.label, pb.label, "error").Inc()
                        metrics.ErrorCount.WithLabelValues(
                                pb.label, "", "put error").Inc()
                        return err
                }
        }
        return nil
}

// NullAnnotator satisfies the Annotatable interface without actually doing
// anything.
type NullAnnotator struct{}

// GetLogTime returns current time rather than the actual row time.
func (row *NullAnnotator) GetLogTime() time.Time {
        return time.Now()
}

// GetClientIPs returns an empty array so nothing is annotated.
func (row *NullAnnotator) GetClientIPs() []string {
        return []string{}
}

// GetServerIP returns an empty string because there is nothing to annotate.
func (row *NullAnnotator) GetServerIP() string {
        return ""
}

// AnnotateClients does nothing.
func (row *NullAnnotator) AnnotateClients(map[string]*api.Annotations) error {
        return nil
}

// AnnotateServer does nothing.
func (row *NullAnnotator) AnnotateServer(*api.Annotations) error {
        return nil
}

1	package row
2
3	// TODO integrate this functionality into the go code.
4	// Probably should have Base implement Parser.
5
6	import (
7	"context"
8	"errors"
9	"io"
10	"log"
11	"sync"
12	"time"
13
14	"github.com/prometheus/client_golang/prometheus"
15
16	"github.com/m-lab/annotation-service/api"
17	v2as "github.com/m-lab/annotation-service/api/v2"
18	"github.com/m-lab/go/logx"
19
20	"github.com/m-lab/etl/metrics"
21	)
22
23	// Errors that may be returned by Buffer functions.
24	var (
25	ErrAnnotationError = errors.New("Annotation error")
26	ErrNotAnnotatable = errors.New("object does not implement Annotatable")
27	ErrBufferFull = errors.New("Buffer full")
28	ErrInvalidSink = errors.New("Not a valid row.Sink")
29	)
30
31	// Annotatable interface enables integration of annotation into parser.Base.
32	// The row type should implement the interface, and the annotations will be added
33	// prior to insertion.
34	type Annotatable interface {
35	GetLogTime() time.Time
36	GetClientIPs() []string // This is a slice to support mutliple hops in traceroute data.
37	GetServerIP() string
38	AnnotateClients(map[string]*api.Annotations) error // Must properly handle missing annotations.
39	AnnotateServer(*api.Annotations) error // Must properly handle nil parameter.
40	}
41
42	// Stats contains stats about buffer history.
43	type Stats struct {
44	Buffered int // rows buffered but not yet sent.
45	Pending int // pending counts previously buffered rows that are being committed.
46	Committed int
47	Failed int
48	}
49
50	// Total returns the total number of rows handled.
51	func (s Stats) Total() int {	×
52	return s.Buffered + s.Pending + s.Committed + s.Failed	×
53	}	×
54
55	// ActiveStats is a stats object that supports updates.
56	type ActiveStats struct {
57	lock sync.RWMutex // Protects all Stats fields.
58	Stats
59	}
60
61	// GetStats implements HasStats()
62	func (as *ActiveStats) GetStats() Stats {	1✔
63	as.lock.RLock()	1✔
64	defer as.lock.RUnlock()	1✔
65	return as.Stats	1✔
66	}	1✔
67
68	// MoveToPending increments the Pending field.
69	func (as *ActiveStats) MoveToPending(n int) {	1✔
70	as.lock.Lock()	1✔
71	defer as.lock.Unlock()	1✔
72	as.Buffered -= n	1✔
73	if as.Buffered < 0 {	1✔
74	log.Println("BROKEN - negative buffered")	×
75	}	×
76	as.Pending += n	1✔
77	}
78
79	// Inc increments the Buffered field
80	func (as *ActiveStats) Inc() {	1✔
81	logx.Debug.Println("IncPending")	1✔
82	as.lock.Lock()	1✔
83	defer as.lock.Unlock()	1✔
84	as.Buffered++	1✔
85	}	1✔
86
87	// Done updates the pending to failed or committed.
88	func (as *ActiveStats) Done(n int, err error) {	1✔
89	as.lock.Lock()	1✔
90	defer as.lock.Unlock()	1✔
91	as.Pending -= n	1✔
92	if as.Pending < 0 {	1✔
93	log.Println("BROKEN: negative Pending")	×
94	}	×
95	if err != nil {	1✔
96	as.Failed += n	×
97	} else {	1✔
98	as.Committed += n	1✔
99	}	1✔
100	logx.Debug.Printf("Done %d->%d %v\n", as.Pending+n, as.Pending, err)	1✔
101	}
102
103	// HasStats can provide stats
104	type HasStats interface {
105	GetStats() Stats
106	}
107
108	// Sink defines the interface for committing rows.
109	// Returns the number of rows successfully committed, and error.
110	// Implementations should be threadsafe.
111	type Sink interface {
112	Commit(rows []interface{}, label string) (int, error)
113	io.Closer
114	}
115
116	// Buffer provides all basic functionality generally needed for buffering, annotating, and inserting
117	// rows that implement Annotatable.
118	// Buffer functions are THREAD-SAFE
119	type Buffer struct {
120	lock sync.Mutex
121	size int // Number of rows before starting new buffer.
122	rows []interface{}
123	}
124
125	// NewBuffer returns a new buffer of the desired size.
126	func NewBuffer(size int) *Buffer {	1✔
127	return &Buffer{size: size, rows: make([]interface{}, 0, size)}	1✔
128	}	1✔
129
130	// Append appends a row to the buffer.
131	// If buffer is full, this returns the buffered rows, and saves provided row
132	// in new buffer. Client MUST handle the returned rows.
133	func (buf *Buffer) Append(row interface{}) []interface{} {	1✔
134	buf.lock.Lock()	1✔
135	defer buf.lock.Unlock()	1✔
136	if len(buf.rows) < buf.size {	2✔
137	buf.rows = append(buf.rows, row)	1✔
138	return nil	1✔
139	}	1✔
140	rows := buf.rows	1✔
141	buf.rows = make([]interface{}, 0, buf.size)	1✔
142	buf.rows = append(buf.rows, row)	1✔
143		1✔
144	return rows	1✔
145	}
146
147	// Reset clears the buffer, returning all pending rows.
148	func (buf *Buffer) Reset() []interface{} {	1✔
149	buf.lock.Lock()	1✔
150	defer buf.lock.Unlock()	1✔
151	res := buf.rows	1✔
152	buf.rows = make([]interface{}, 0, buf.size)	1✔
153	return res	1✔
154	}	1✔
155
156	type annotator struct {
157	v2 v2as.Annotator
158	}
159
160	// label is used to label metrics and errors in GetAnnotations
161	func (ann *annotator) annotateServers(rows []interface{}, label string) error {	1✔
162	serverIPs := make(map[string]struct{})	1✔
163	logTime := time.Time{}	1✔
164	for i := range rows {	2✔
165	r, ok := rows[i].(Annotatable)	1✔
166	if !ok {	1✔
167	return ErrNotAnnotatable	×
168	}	×
169
170	// Only ask for the IP if it is non-empty.
171	ip := r.GetServerIP()	1✔
172	if ip != "" {	2✔
173	serverIPs[ip] = struct{}{}	1✔
174	}	1✔
175
176	if (logTime == time.Time{}) {	2✔
177	logTime = r.GetLogTime()	1✔
178	}	1✔
179	}
180
181	ipSlice := make([]string, 0, len(rows))	1✔
182	for ip := range serverIPs {	2✔
183	ipSlice = append(ipSlice, ip)	1✔
184	}	1✔
185	if len(ipSlice) == 0 {	1✔
186	return nil	×
187	}	×
188	response, err := ann.v2.GetAnnotations(context.Background(), logTime, ipSlice, label)	1✔
189	if err != nil {	1✔
190	log.Println("error in server GetAnnotations: ", err)	×
191	metrics.AnnotationErrorCount.With(prometheus.	×
192	Labels{"source": "Server IP: RPC err in GetAnnotations."}).Inc()	×
193	return err	×
194	}	×
195	annMap := response.Annotations	1✔
196	if annMap == nil {	1✔
197	log.Println("empty server annotation response")	×
198	metrics.AnnotationErrorCount.With(prometheus.	×
199	Labels{"source": "Server IP: empty response"}).Inc()	×
200	return ErrAnnotationError	×
201	}	×
202
203	for i := range rows {	2✔
204	r, ok := rows[i].(Annotatable)	1✔
205	if !ok {	1✔
206	err = ErrNotAnnotatable	×
207	} else {	1✔
208	ip := r.GetServerIP()	1✔
209	if ip != "" {	2✔
210	ann, ok := annMap[ip]	1✔
211	if ok {	2✔
212	r.AnnotateServer(ann)	1✔
213	}	1✔
214	}
215	}
216	}
217
218	return err	1✔
219	}
220
221	var logEmptyAnn = logx.NewLogEvery(nil, 60*time.Second)
222
223	// label is used to label metrics and errors in GetAnnotations
224	func (ann *annotator) annotateClients(rows []interface{}, label string) error {	1✔
225	ipSlice := make([]string, 0, 2*len(rows)) // This may be inadequate, but its a reasonable start.	1✔
226	logTime := time.Time{}	1✔
227	for i := range rows {	2✔
228	r, ok := rows[i].(Annotatable)	1✔
229	if !ok {	1✔
230	return ErrNotAnnotatable	×
231	}	×
232	ipSlice = append(ipSlice, r.GetClientIPs()...)	1✔
233	if (logTime == time.Time{}) {	2✔
234	logTime = r.GetLogTime()	1✔
235	}	1✔
236	}
237
238	response, err := ann.v2.GetAnnotations(context.Background(), logTime, ipSlice, label)	1✔
239	if err != nil {	1✔
240	log.Println("error in client GetAnnotations: ", err)	×
241	metrics.AnnotationErrorCount.With(prometheus.	×
242	Labels{"source": "Client IP: RPC err in GetAnnotations."}).Inc()	×
243	return err	×
244	}	×
245	annMap := response.Annotations	1✔
246	if annMap == nil {	1✔
247	logEmptyAnn.Println("empty client annotation response")	×
248	metrics.AnnotationErrorCount.With(prometheus.	×
249	Labels{"source": "Client IP: empty response"}).Inc()	×
250	return ErrAnnotationError	×
251	}	×
252
253	for i := range rows {	2✔
254	r, ok := rows[i].(Annotatable)	1✔
255	if !ok {	1✔
256	err = ErrNotAnnotatable	×
257	} else {	1✔
258	// Will not error because we check for nil annMap above.	1✔
259	r.AnnotateClients(annMap)	1✔
260	}	1✔
261	}
262
263	return err	1✔
264	}
265
266	// Annotate fetches and applies annotations for all rows
267	func (ann *annotator) Annotate(rows []interface{}, metricLabel string) error {	1✔
268	metrics.WorkerState.WithLabelValues(metricLabel, "annotate").Inc()	1✔
269	defer metrics.WorkerState.WithLabelValues(metricLabel, "annotate").Dec()	1✔
270	if len(rows) == 0 {	1✔
271	return nil	×
272	}	×
273	// TODO replace this with a histogram.
274	defer func(label string, start time.Time) {	2✔
275	metrics.AnnotationTimeSummary.With(prometheus.Labels{"test_type": label}).Observe(float64(time.Since(start).Nanoseconds()))	1✔
276	}(metricLabel, time.Now())	1✔
277
278	// TODO Consider doing these in parallel?
279	clientErr := ann.annotateClients(rows, metricLabel)	1✔
280	serverErr := ann.annotateServers(rows, metricLabel)	1✔
281		1✔
282	if clientErr != nil {	1✔
283	return clientErr	×
284	}	×
285
286	if serverErr != nil {	1✔
287	return serverErr	×
288	}	×
289
290	return nil	1✔
291	}
292
293	// Base provides common parser functionality.
294	// Base is NOT THREAD-SAFE
295	type Base struct {
296	sink Sink
297	ann annotator
298	buf *Buffer
299	label string // Used in metrics and errors.
300
301	stats ActiveStats
302	}
303
304	// NewBase creates a new Base. This will generally be embedded in a type specific parser.
305	func NewBase(label string, sink Sink, bufSize int, ann v2as.Annotator) *Base {	1✔
306	buf := NewBuffer(bufSize)	1✔
307	return &Base{sink: sink, ann: annotator{ann}, buf: buf, label: label}	1✔
308	}	1✔
309
310	// GetStats returns the buffer/sink stats.
311	func (pb *Base) GetStats() Stats {	1✔
312	return pb.stats.GetStats()	1✔
313	}	1✔
314
315	// TaskError return the task level error, based on failed rows, or any other criteria.
316	func (pb *Base) TaskError() error {	×
317	return nil	×
318	}	×
319
320	var logAnnError = logx.NewLogEvery(nil, 60*time.Second)
321
322	func (pb *Base) commit(rows []interface{}) error {	1✔
323	err := pb.ann.Annotate(rows, pb.label)	1✔
324	if err != nil {	1✔
325	logAnnError.Println("annotation: ", err)	×
326	}	×
327
328	// TODO do we need these to be done in order.
329	// This is synchronous, blocking, and thread safe.
330	done, err := pb.sink.Commit(rows, pb.label)	1✔
331	if done > 0 {	2✔
332	pb.stats.Done(done, nil)	1✔
333	}	1✔
334	if err != nil {	1✔
335	log.Println(pb.label, err)	×
336	pb.stats.Done(len(rows)-done, err)	×
337	}	×
338	return err	1✔
339	}
340
341	// Flush synchronously flushes any pending rows.
342	func (pb *Base) Flush() error {	1✔
343	rows := pb.buf.Reset()	1✔
344	pb.stats.MoveToPending(len(rows))	1✔
345	return pb.commit(rows)	1✔
346	}	1✔
347
348	// Put adds a row to the buffer.
349	// Iff the buffer is already full the prior buffered rows are
350	// annotated and committed to the Sink.
351	// NOTE: There is no guarantee about ordering of writes resulting from
352	// sequential calls to Put. However, once a block of rows is submitted
353	// to pb.commit, it should be written in the same order to the Sink.
354	// TODO improve Annotatable architecture.
355	func (pb *Base) Put(row Annotatable) error {	1✔
356	rows := pb.buf.Append(row)	1✔
357	pb.stats.Inc()	1✔
358		1✔
359	if rows != nil {	2✔
360	pb.stats.MoveToPending(len(rows))	1✔
361	err := pb.commit(rows)	1✔
362	if err != nil {	1✔
363	// Note that error is likely associated with buffered rows, not the current	×
364	// row.	×
365	// When using GCS output, this may result in a corrupted json file.	×
366	// In that event, the test count may become meaningless.	×
367	metrics.TestTotal.WithLabelValues(pb.label, pb.label, "error").Inc()	×
368	metrics.ErrorCount.WithLabelValues(	×
369	pb.label, "", "put error").Inc()	×
370	return err	×
371	}	×
372	}
373	return nil	1✔
374	}
375
376	// NullAnnotator satisfies the Annotatable interface without actually doing
377	// anything.
378	type NullAnnotator struct{}
379
380	// GetLogTime returns current time rather than the actual row time.
381	func (row *NullAnnotator) GetLogTime() time.Time {	×
382	return time.Now()	×
383	}	×
384
385	// GetClientIPs returns an empty array so nothing is annotated.
386	func (row *NullAnnotator) GetClientIPs() []string {	×
387	return []string{}	×
388	}	×
389
390	// GetServerIP returns an empty string because there is nothing to annotate.
391	func (row *NullAnnotator) GetServerIP() string {	×
392	return ""	×
393	}	×
394
395	// AnnotateClients does nothing.
396	func (row NullAnnotator) AnnotateClients(map[string]api.Annotations) error {	×
397	return nil	×
398	}	×
399
400	// AnnotateServer does nothing.
401	func (row NullAnnotator) AnnotateServer(api.Annotations) error {	×
402	return nil	×
403	}	×

m-lab / etl / 7189

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