7224

Committed 03 Mar 2022 06:38PM UTC coverage: 64.587% (+0.02%) from 64.563%

Build # 7224

Build Type

push

travis-ci-com

Committed by

cristinaleonr

Commit Message

<a href="https://github.com/m-lab/etl/commit/<a class=hub.com/m-lab/etl/commit/5ca4adf80051e0601a8cbff25a38b578e11f1e9f">5ca4adf80<a href="https://github.com/m-lab/etl/commit/5ca4adf80051e0601a8cbff25a38b578e11f1e9f">&quot;&gt;Revert &amp;quot;Return combined error&quot;

This reverts commit </a><a class="double-link" href="https://github.com/m-lab/etl/commit/<a class="double-link" href="https://github.com/m-lab/etl/commit/7b67ce2ebe841b4c0e85d2690feb2333472493c6">7b67ce2eb</a>">7b67ce2eb</a><a href="https://github.com/m-lab/etl/commit/5ca4adf80051e0601a8cbff25a38b578e11f1e9f">.

Run Details

9 of 9 new or added lines in 2 files covered. (100.0%)

3985 of 6170 relevant lines covered (64.59%)

0.7 hits per line

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

72.65

/task/task.go

// Package task provides the tracking of state for a single task pushed by the
// external task queue.
//
// The Task type ...
// TODO(dev) Improve comments and header before merging to dev.
package task

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

        "cloud.google.com/go/bigquery"

        "github.com/m-lab/etl/etl"
        "github.com/m-lab/etl/metrics"
        "github.com/m-lab/etl/row"
        "github.com/m-lab/etl/storage"
        "github.com/m-lab/go/logx"
)

// Factory provides Get() which always returns a new, complete Task.
// TODO for the defs that stay in factory package, remove ...Factory.
type Factory interface {
        Get(context.Context, etl.DataPath) (*Task, etl.ProcessingError)
}

// DefaultMaxFileSize is the default value for the file size limit for calls to NextTest.
// Larger than this risks an OOM if there are
// multiple large files at on multiple tasks.
// This can be overridden with SetMaxFileSize()
const DefaultMaxFileSize = 200 * 1024 * 1024

// Task contains the state required to process a single task tar file.
// TODO(dev) Add unit tests for meta data.
type Task struct {
        // TestSource and Parser are both embedded, so their interfaces are delegated
        // to the component structs.
        etl.TestSource // Source from which to read tests.
        etl.Parser     // Parser to parse the tests.

        meta        map[string]bigquery.Value // Metadata about this task.
        maxFileSize int64                     // Max file size to avoid OOM.

        closer io.Closer // So we can call Close()
}

// NewTask constructs a task, injecting the source and the parser.
func NewTask(filename string, src etl.TestSource, prsr etl.Parser, closer io.Closer) *Task {
        // TODO - should the meta data be a nested type?
        meta := make(map[string]bigquery.Value, 3)
        meta["filename"] = filename
        meta["parse_time"] = time.Now()
        meta["attempt"] = 1
        meta["date"] = src.Date()
        t := Task{
                TestSource:  src,
                Parser:      prsr,
                meta:        meta,
                maxFileSize: DefaultMaxFileSize,
                closer:      closer}
        return &t
}

// Close closes the source and sink.
func (tt *Task) Close() {
        tt.TestSource.Close()
        tt.closer.Close()
}

// SetMaxFileSize overrides the default maxFileSize.
func (tt *Task) SetMaxFileSize(max int64) {
        tt.maxFileSize = max
}

// This is used for logging empty test warnings.
// TODO - consider just removing the log.
var emptyTest = logx.NewLogEvery(nil, time.Second)

// ProcessAllTests loops through all the tests in a tar file, calls the
// injected parser to parse them, and inserts them into bigquery. Returns the
// number of files processed.
// TODO pass in the datatype label.
func (tt *Task) ProcessAllTests(failfast bool) (int, error) {
        if tt.Parser == nil {
                panic("Parser is nil")
        }
        metrics.WorkerState.WithLabelValues(tt.Type(), "task").Inc()
        defer metrics.WorkerState.WithLabelValues(tt.Type(), "task").Dec()
        files := 0
        nilData := 0
        var testname string
        var data []byte
        var loopErr error
        // Read each file from the tar

OUTER:
        for testname, data, loopErr = tt.NextTest(tt.maxFileSize); loopErr != io.EOF; testname, data, loopErr = tt.NextTest(tt.maxFileSize) {
                files++
                if loopErr != nil {
                        switch {
                        case loopErr == io.EOF:
                                break OUTER
                        case loopErr == storage.ErrOversizeFile:
                                log.Printf("ERROR filename:%s testname:%s files:%d, duration:%v err:%v",
                                        tt.meta["filename"], testname, files,
                                        time.Since(tt.meta["parse_time"].(time.Time)), loopErr)
                                metrics.TestTotal.WithLabelValues(
                                        tt.Type(), "unknown", "oversize file").Inc()
                                continue OUTER
                        default:
                                // We are seeing several of these per hour, a little more than
                                // one in one thousand files.  duration varies from 10 seconds
                                // up to several minutes.
                                // Example:
                                // filename:
                                // gs://m-lab-sandbox/ndt/2016/04/10/20160410T000000Z-mlab1-ord02-ndt-0002.tgz
                                // files:666 duration:1m47.571825351s
                                // err:stream error: stream ID 801; INTERNAL_ERROR
                                // Because of the break, this error is passed up, and counted at
                                // the Task level.
                                log.Printf("ERROR filename:%s testname:%s files:%d, duration:%v err:%v",
                                        tt.meta["filename"], testname, files,
                                        time.Since(tt.meta["parse_time"].(time.Time)), loopErr)

                                metrics.TestTotal.WithLabelValues(
                                        tt.Type(), "unknown", "unrecovered").Inc()
                                // Since we don't understand these errors, safest thing to do is
                                // stop processing the tar file (and task).
                                break OUTER
                        }
                }
                if data == nil {
                        // TODO(dev) Handle directories (expected) and other
                        // things separately.
                        nilData++
                        // If verbose, log the filename that is skipped.
                        continue
                }
                if len(data) == 0 {
                        // Parser should also, likely, insert an empty row with just parse info and id
                        // There are spike of 100K, so we use a 1 second logEvery to avoid log spam.
                        emptyTest.Printf("WARNING empty test %s:%s %s\n", tt.TableName(), tt.Type(), tt.Detail())
                        metrics.WarningCount.WithLabelValues(
                                tt.TableName(), tt.Type(), "empty test file").Inc()
                }
                kind, parsable := tt.Parser.IsParsable(testname, data)
                if !parsable {
                        metrics.FileSizeHistogram.WithLabelValues(
                                tt.Type(), kind, "ignored").Observe(float64(len(data)))
                        // Don't bother calling ParseAndInsert since this is unparsable.
                        continue
                } else {
                        metrics.FileSizeHistogram.WithLabelValues(
                                tt.Type(), kind, "parsed").Observe(float64(len(data)))
                }
                loopErr = tt.Parser.ParseAndInsert(tt.meta, testname, data)
                // Shouldn't have any of these, as they should be handled in ParseAndInsert.
                if loopErr != nil {
                        log.Printf("ERROR %v", loopErr)
                        // TODO(dev) Handle this error properly!
                        commitRowErr := row.ErrCommitRow{}
                        if failfast && errors.As(loopErr, &commitRowErr) {
                                break OUTER
                        }
                        continue
                }
        }

        // There may be an error from the processing loop, but we wait to handle that
        // error until after we flush and cached rows.
        flushErr := tt.Flush()
        if flushErr != nil {
                log.Printf("%v", flushErr)
        }

        // TODO - make this debug or remove
        log.Printf("Processed %d files, %d nil data, %d rows committed, %d failed, from %s into %s",
                files, nilData, tt.Parser.Committed(), tt.Parser.Failed(),
                tt.meta["filename"], tt.Parser.FullTableName())

        // We expect the loopErr to be io.EOF.  If it is something else, then
        // it is an actual error, and we want to return that error.
        if loopErr != io.EOF {
                return files, loopErr
        }

        // Check if the overall task is OK, or should be rejected.
        if tt.Parser.TaskError() != nil {
                return files, tt.Parser.TaskError()
        }
        // Otherwise, return any error from the call to Flush.
        return files, flushErr
}

1	// Package task provides the tracking of state for a single task pushed by the
2	// external task queue.
3	//
4	// The Task type ...
5	// TODO(dev) Improve comments and header before merging to dev.
6	package task
7
8	import (
9	"context"
10	"errors"
11	"io"
12	"log"
13	"time"
14
15	"cloud.google.com/go/bigquery"
16
17	"github.com/m-lab/etl/etl"
18	"github.com/m-lab/etl/metrics"
19	"github.com/m-lab/etl/row"
20	"github.com/m-lab/etl/storage"
21	"github.com/m-lab/go/logx"
22	)
23
24	// Factory provides Get() which always returns a new, complete Task.
25	// TODO for the defs that stay in factory package, remove ...Factory.
26	type Factory interface {
27	Get(context.Context, etl.DataPath) (*Task, etl.ProcessingError)
28	}
29
30	// DefaultMaxFileSize is the default value for the file size limit for calls to NextTest.
31	// Larger than this risks an OOM if there are
32	// multiple large files at on multiple tasks.
33	// This can be overridden with SetMaxFileSize()
34	const DefaultMaxFileSize = 200 * 1024 * 1024
35
36	// Task contains the state required to process a single task tar file.
37	// TODO(dev) Add unit tests for meta data.
38	type Task struct {
39	// TestSource and Parser are both embedded, so their interfaces are delegated
40	// to the component structs.
41	etl.TestSource // Source from which to read tests.
42	etl.Parser // Parser to parse the tests.
43
44	meta map[string]bigquery.Value // Metadata about this task.
45	maxFileSize int64 // Max file size to avoid OOM.
46
47	closer io.Closer // So we can call Close()
48	}
49
50	// NewTask constructs a task, injecting the source and the parser.
51	func NewTask(filename string, src etl.TestSource, prsr etl.Parser, closer io.Closer) *Task {	1✔
52	// TODO - should the meta data be a nested type?	1✔
53	meta := make(map[string]bigquery.Value, 3)	1✔
54	meta["filename"] = filename	1✔
55	meta["parse_time"] = time.Now()	1✔
56	meta["attempt"] = 1	1✔
57	meta["date"] = src.Date()	1✔
58	t := Task{	1✔
59	TestSource: src,	1✔
60	Parser: prsr,	1✔
61	meta: meta,	1✔
62	maxFileSize: DefaultMaxFileSize,	1✔
63	closer: closer}	1✔
64	return &t	1✔
65	}	1✔
66
67	// Close closes the source and sink.
68	func (tt *Task) Close() {	×
69	tt.TestSource.Close()	×
70	tt.closer.Close()	×
71	}	×
72
73	// SetMaxFileSize overrides the default maxFileSize.
74	func (tt *Task) SetMaxFileSize(max int64) {	1✔
75	tt.maxFileSize = max	1✔
76	}	1✔
77
78	// This is used for logging empty test warnings.
79	// TODO - consider just removing the log.
80	var emptyTest = logx.NewLogEvery(nil, time.Second)
81
82	// ProcessAllTests loops through all the tests in a tar file, calls the
83	// injected parser to parse them, and inserts them into bigquery. Returns the
84	// number of files processed.
85	// TODO pass in the datatype label.
86	func (tt *Task) ProcessAllTests(failfast bool) (int, error) {	1✔
87	if tt.Parser == nil {	1✔
88	panic("Parser is nil")	×
89	}
90	metrics.WorkerState.WithLabelValues(tt.Type(), "task").Inc()	1✔
91	defer metrics.WorkerState.WithLabelValues(tt.Type(), "task").Dec()	1✔
92	files := 0	1✔
93	nilData := 0	1✔
94	var testname string	1✔
95	var data []byte	1✔
96	var loopErr error	1✔
97	// Read each file from the tar	1✔
98		1✔
99	OUTER:	1✔
100	for testname, data, loopErr = tt.NextTest(tt.maxFileSize); loopErr != io.EOF; testname, data, loopErr = tt.NextTest(tt.maxFileSize) {	2✔
101	files++	1✔
102	if loopErr != nil {	2✔
103	switch {	1✔
104	case loopErr == io.EOF:	×
105	break OUTER	×
106	case loopErr == storage.ErrOversizeFile:	1✔
107	log.Printf("ERROR filename:%s testname:%s files:%d, duration:%v err:%v",	1✔
108	tt.meta["filename"], testname, files,	1✔
109	time.Since(tt.meta["parse_time"].(time.Time)), loopErr)	1✔
110	metrics.TestTotal.WithLabelValues(	1✔
111	tt.Type(), "unknown", "oversize file").Inc()	1✔
112	continue OUTER	1✔
113	default:	1✔
114	// We are seeing several of these per hour, a little more than	1✔
115	// one in one thousand files. duration varies from 10 seconds	1✔
116	// up to several minutes.	1✔
117	// Example:	1✔
118	// filename:	1✔
119	// gs://m-lab-sandbox/ndt/2016/04/10/20160410T000000Z-mlab1-ord02-ndt-0002.tgz	1✔
120	// files:666 duration:1m47.571825351s	1✔
121	// err:stream error: stream ID 801; INTERNAL_ERROR	1✔
122	// Because of the break, this error is passed up, and counted at	1✔
123	// the Task level.	1✔
124	log.Printf("ERROR filename:%s testname:%s files:%d, duration:%v err:%v",	1✔
125	tt.meta["filename"], testname, files,	1✔
126	time.Since(tt.meta["parse_time"].(time.Time)), loopErr)	1✔
127		1✔
128	metrics.TestTotal.WithLabelValues(	1✔
129	tt.Type(), "unknown", "unrecovered").Inc()	1✔
130	// Since we don't understand these errors, safest thing to do is	1✔
131	// stop processing the tar file (and task).	1✔
132	break OUTER	1✔
133	}
134	}
135	if data == nil {	1✔
136	// TODO(dev) Handle directories (expected) and other	×
137	// things separately.	×
138	nilData++	×
139	// If verbose, log the filename that is skipped.	×
140	continue	×
141	}
142	if len(data) == 0 {	1✔
143	// Parser should also, likely, insert an empty row with just parse info and id	×
144	// There are spike of 100K, so we use a 1 second logEvery to avoid log spam.	×
145	emptyTest.Printf("WARNING empty test %s:%s %s\n", tt.TableName(), tt.Type(), tt.Detail())	×
146	metrics.WarningCount.WithLabelValues(	×
147	tt.TableName(), tt.Type(), "empty test file").Inc()	×
148	}	×
149	kind, parsable := tt.Parser.IsParsable(testname, data)	1✔
150	if !parsable {	1✔
151	metrics.FileSizeHistogram.WithLabelValues(	×
152	tt.Type(), kind, "ignored").Observe(float64(len(data)))	×
153	// Don't bother calling ParseAndInsert since this is unparsable.	×
154	continue	×
155	} else {	1✔
156	metrics.FileSizeHistogram.WithLabelValues(	1✔
157	tt.Type(), kind, "parsed").Observe(float64(len(data)))	1✔
158	}	1✔
159	loopErr = tt.Parser.ParseAndInsert(tt.meta, testname, data)	1✔
160	// Shouldn't have any of these, as they should be handled in ParseAndInsert.	1✔
161	if loopErr != nil {	1✔
162	log.Printf("ERROR %v", loopErr)	×
163	// TODO(dev) Handle this error properly!	×
164	commitRowErr := row.ErrCommitRow{}	×
165	if failfast && errors.As(loopErr, &commitRowErr) {	×
166	break OUTER	×
167	}
168	continue	×
169	}
170	}
171
172	// There may be an error from the processing loop, but we wait to handle that
173	// error until after we flush and cached rows.
174	flushErr := tt.Flush()	1✔
175	if flushErr != nil {	1✔
176	log.Printf("%v", flushErr)	×
177	}	×
178
179	// TODO - make this debug or remove
180	log.Printf("Processed %d files, %d nil data, %d rows committed, %d failed, from %s into %s",	1✔
181	files, nilData, tt.Parser.Committed(), tt.Parser.Failed(),	1✔
182	tt.meta["filename"], tt.Parser.FullTableName())	1✔
183		1✔
184	// We expect the loopErr to be io.EOF. If it is something else, then	1✔
185	// it is an actual error, and we want to return that error.	1✔
186	if loopErr != io.EOF {	2✔
187	return files, loopErr	1✔
188	}	1✔
189
190	// Check if the overall task is OK, or should be rejected.
191	if tt.Parser.TaskError() != nil {	1✔
192	return files, tt.Parser.TaskError()	×
193	}	×
194	// Otherwise, return any error from the call to Flush.
195	return files, flushErr	1✔
196	}

m-lab / etl / 7224

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