From 4e0a844cfb57d3b2ae64581db54265f7f5510e31 Mon Sep 17 00:00:00 2001 From: "jonathan.cheng" Date: Thu, 6 Oct 2022 14:20:35 -0700 Subject: [PATCH] Sync from internal repo through 6035345 --- idk/Makefile | 4 +- idk/cmd/molecula-consumer-kinesis/main.go | 20 +++++--- idk/ingest.go | 36 ++++++++++++-- idk/ingest_test.go | 58 +++++++++++++++++++++++ 4 files changed, 106 insertions(+), 12 deletions(-) diff --git a/idk/Makefile b/idk/Makefile index 4d011d516..80bae7b61 100644 --- a/idk/Makefile +++ b/idk/Makefile @@ -227,11 +227,11 @@ test-run: testenv vendor test-run-race: testenv vendor $(DOCKER_COMPOSE) build idk-test - $(DOCKER_COMPOSE) run -T idk-test bash -c "set -o pipefail; go test -v -mod=vendor -race -covermode=atomic -tags=dynamic $(TPKG) -coverpkg=$(TPKG) -timeout=30m -json -coverprofile=/testdata/$(PROJECT)_coverage.out | tee /testdata/$(PROJECT)_report.out" + $(DOCKER_COMPOSE) run -T idk-test bash -c "set -o pipefail; go test -v -mod=vendor -race -covermode=atomic -tags=dynamic $(TPKG) -coverpkg=$(TPKG) -timeout=30m -json -coverprofile=/testdata/$(PROJECT)_race_coverage.out | tee /testdata/$(PROJECT)_report.out" test-run-kafka-sasl: testenv vendor $(DOCKER_COMPOSE) build idk-test - $(DOCKER_COMPOSE) run -T idk-test bash -c "set -o pipefail; go test -v --tags=kafka_sasl -mod=vendor -race -timeout=30m $(TPKG) -covermode=atomic -coverpkg=$(TPKG) -json -coverprofile=/testdata/$(PROJECT)_coverage.out | tee /testdata/$(PROJECT)_report.out" + $(DOCKER_COMPOSE) run -T idk-test bash -c "set -o pipefail; go test -v --tags=kafka_sasl -mod=vendor -race -timeout=30m $(TPKG) -covermode=atomic -coverpkg=$(TPKG) -json -coverprofile=/testdata/$(PROJECT)_sasl_coverage.out | tee /testdata/$(PROJECT)_report.out" .pulled: $(DOCKER_COMPOSE) pull diff --git a/idk/cmd/molecula-consumer-kinesis/main.go b/idk/cmd/molecula-consumer-kinesis/main.go index d25ca7c0e..f37572d6f 100644 --- a/idk/cmd/molecula-consumer-kinesis/main.go +++ b/idk/cmd/molecula-consumer-kinesis/main.go @@ -9,18 +9,24 @@ import ( "github.com/featurebasedb/featurebase/v3/logger" ) -func logFailure(errorType kinesis.ErrorType, m *kinesis.Main, v interface{}) { +func logError(m *kinesis.Main, err error) { log := m.Log() if log == nil { log = logger.NewStandardLogger(os.Stderr) } + log.Errorf("Error running command: %s", err) - if errorType == kinesis.RecoverableErrorType { - log.Errorf("Error running command: %+v", v) - } else { - log.Panicf("Panic running command: %+v", v) +} + +func logPanic(m *kinesis.Main, v interface{}) { + log := m.Log() + + if log == nil { + log = logger.NewStandardLogger(os.Stderr) } + log.Panicf("Panic running command: %+v", v) + } func main() { @@ -33,7 +39,7 @@ func main() { // Capture any panic and log it before dying. defer func() { if r := recover(); r != nil { - logFailure(kinesis.PanicErrorType, m, r) + logPanic(m, r) os.Exit(1) } }() @@ -44,7 +50,7 @@ func main() { } if err := m.Run(); err != nil { - logFailure(kinesis.RecoverableErrorType, m, err) + logError(m, err) os.Exit(1) } } diff --git a/idk/ingest.go b/idk/ingest.go index 7722bdcab..f359f0c09 100644 --- a/idk/ingest.go +++ b/idk/ingest.go @@ -92,6 +92,7 @@ type Main struct { AllowIntOutOfRange bool `help:"Allow ingest to continue when it encounters out of range integers in IntFields. (default false)"` AllowDecimalOutOfRange bool `help:"Allow ingest to continue when it encounters out of range decimals in DecimalFields. (default false)"` AllowTimestampOutOfRange bool `help:"Allow ingest to continue when it encounters out of range timestamps in TimestampFields. (default false)"` + SkipBadRows int `help:"If you fail to process the first n rows without processing one successfully, fail."` UseShardTransactionalEndpoint bool `flag:"use-shard-transactional-endpoint" help:"Use alternate import endpoint. Currently unstable/testing"` @@ -325,6 +326,8 @@ func (m *Main) ingest(ctx context.Context, source Source, nexter IDAllocator, so var recordizers []Recordizer var prevRec Record var row *pilosaclient.Row + var errorCounter int // keeps track of consecuitive errors across records + var anyRecordSuccessful bool if m.progress != nil { source = m.progress.Track(source) } @@ -480,6 +483,7 @@ initialFetch: } } + rowHasError := false for _, rdz := range recordizers { err = rdz(data, row) if err != nil { @@ -487,11 +491,32 @@ initialFetch: // excludes 'out of range' errors so that ingest can continue importing the rest // of the records if !m.allowError(err) { - return err + rowHasError = true + // must return error and exit idk when SkipBadRows is not defined (or set to 0) + if m.SkipBadRows == 0 { + return err + } else { + // will handle this error based on errorCounter in the later if block + m.Log().Errorf("Bad record: +%v, reason: %v\n", row, err) + break + } } m.Log().Errorf(err.Error()) } } + + if !anyRecordSuccessful && rowHasError { + // We cannot allow a certain number of consecutive errors in the beginning of ingest. + errorCounter++ + if errorCounter > m.SkipBadRows { + return errors.Wrapf(err, "consecutive bad records exceeded limit, errorCounter: %d\n", errorCounter) // wraps the current recordizer error + } + err = nil // already logged the recordizer error, no need to propagate the err + } else { + anyRecordSuccessful = true // after this is set true, we will skip any bad rows in the future + err = nil + } + if nexter != nil { // add ID if no id field specified if batchStart { rerr := nexter.Reserve(ctx, uint64(m.BatchSize)) @@ -522,8 +547,13 @@ initialFetch: } row.ID = id } - err = batch.Add(*row) - m.stats.Count(MetricIngesterRowsAdded, 1, 1) + + // skip bad rows only + if !rowHasError { + err = batch.Add(*row) + m.stats.Count(MetricIngesterRowsAdded, 1, 1) + } + if err == pilosaclient.ErrBatchNowFull || err == pilosaclient.ErrBatchNowStale { batchLen := batch.Len() err = m.importBatch(batch) diff --git a/idk/ingest_test.go b/idk/ingest_test.go index 0b2d17ecb..ae9eb0005 100644 --- a/idk/ingest_test.go +++ b/idk/ingest_test.go @@ -194,6 +194,64 @@ func TestIngestSignedIntBoolField(t *testing.T) { } } +//The following two tests are used to test the functionality of a new feature in which we skip bad records coming into idk and log them. +//First function checks that we can successfully skip some bad rows coming into idk +//Second function checks whether we get error if we have more bad records than acceptable errors by idk mentioned by SkipBadRows parameter. + +func skipBadRowsTestSource() *testSource { + ts := newTestSource([]Field{StringField{NameVal: "rcid"}, SignedIntBoolKeyField{NameVal: "svals"}}, + [][]interface{}{ + {"c", "badrecord1"}, + {"d", "badrecord2"}, + {"a", "badrecord3"}, + {"x", int64(66)}, + {"b", int64(11)}, + {"b", int64(22)}, + {"b", int64(-32)}, + {"b", int64(-44)}, + {"b", "badrecord4"}, + {"b", int64(11)}, + {"b", int64(7)}, + {"c", int64(5)}, + }) + return ts + +} +func ingesterCreationForSkipBadRowsTest(skipbadrows int) *Main { + ts := skipBadRowsTestSource() + ingester := NewMain() + configureTestFlags(ingester) + ingester.NewSource = func() (Source, error) { return ts, nil } + rand.Seed(time.Now().UTC().UnixNano()) + ingester.Index = fmt.Sprintf("ingestint%d", rand.Intn(100000)) + ingester.BatchSize = 2 + ingester.SkipBadRows = skipbadrows + ingester.PrimaryKeyFields = []string{"rcid"} + return ingester +} +func TestSkipBadRowsFunctionality(t *testing.T) { + + ingester := ingesterCreationForSkipBadRowsTest(3) + + err := ingester.Run() + if err != nil { + t.Fatalf("%s: %v", idktest.ErrRunningIngest, err) + } +} + +func TestSkipBadRowsFunctionalityWhenErrorCountIsMore(t *testing.T) { + + ingester := ingesterCreationForSkipBadRowsTest(1) + + err := ingester.Run() + if err == nil { + t.Fatalf("%s: %v", idktest.ErrRunningIngest, err) + } + if !strings.Contains(err.Error(), "consecutive bad records exceeded limit") { + t.Fatalf("did not receive expected error from idk %v", err.Error()) + } +} + // TestSingleBoolClear essentially creates an import batch which // clears a bit in a particular fragment without setting a bit in that // same fragment. There's a potential optimization in the pilosa client