scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

[CLOUD-934] Optionally broadcast IDK Kinesis errors/panics to externa… (#2185)

Browse source 
* [CLOUD-934] Optionally broadcast IDK Kinesis errors/panics to external storage

- Add a minor public method `idk.Main.SetLog` to allow setting the logger instance
  after initialization.
- Add a Logger implementation that captures recoverable errors and panics
  and pushes to an external store. Meant to decorate an existing Logger
  instance and always delegate to its implementation. Decoration happens
  when all AWS resources are initialized. Before then, the wrapped Logger
  implementation is used.
- If `--error-queue-name/CONSUMER_ERROR_QUEUE_NAME` specified, use an
  ErrorStreamLogger to push errors and panics to an SQS queue with that name.
  Omission of the option preserves current behavior.
- Parse sink ID from the `--stream-name/CONSUMER_STREAM_NAME` expecting the form
  'PREFIX'-VALID_UUID. If the sink UUID is invalid, emit a warning that errors/panics
  will not be written to an SQS queue but will still be logged using the decorated
  Logger instance.
- The inability to push to an SQS queue leads to warnings being emitted to notify
  ECS that no queue will be written to and is NOT a hard error.
- Add SQS interface mock for unit testing.
- Add IDK make targets for generating mock interfaces.

* [CLOUD-934] Execute go mod tidy and go fmt to pass CI/CD checks

* [CLOUD-934] Remove extraneous Makefile in idk/kinesis and fix install-mock-generator target

* [CLOUD-934] Add godocs to exported types and functions

* [CLOUD-934] Changed warning to not sound so ominous and update associated unit test

* [CLOUD-934] Unblock CI/CD at the IDK test stage
This commit is contained in:
wmanongdo 2022-08-12 18:59:56 -04:00 • committed by GitHub
parent 39def696c5
commit baab212bb1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 2561 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.gitlab/.gitlab-ci.yml
View file

@ -139,7 +139,7 @@ build amd container fb:
before_script:
- echo "${DOCKER_DEPLOY_TOKEN}" | docker login -u ${DOCKER_DEPLOY_USER} --password-stdin ${CI_REGISTRY}
script:
- tag=${CI_REGISTRY_IMAGE}/featurebase:linux-amd64-${CI_COMMIT_REF_SLUG}
- tag=${CI_REGISTRY_IMAGE}/featurebase:linux-amd64-${CI_COMMIT_REF_NAME}
- docker build --build-arg GO_VERSION=$GOVERSION --build-arg ARCH=amd64 -t $tag -f .gitlab/Dockerfile .
- docker push $tag
- echo Created docker featurebase image with tag "$tag"
@ -578,7 +578,7 @@ build arm container fb:
before_script:
- echo "${DOCKER_DEPLOY_TOKEN}" | docker login -u ${DOCKER_DEPLOY_USER} --password-stdin ${CI_REGISTRY}
script:
- tag=${CI_REGISTRY_IMAGE}/featurebase:linux-arm64-${CI_COMMIT_REF_SLUG}
- tag=${CI_REGISTRY_IMAGE}/featurebase:linux-arm64-${CI_COMMIT_REF_NAME}
- docker build --build-arg GO_VERSION=$GOVERSION --build-arg ARCH=arm64 -t $tag -f .gitlab/Dockerfile .
- docker push $tag
- echo Created docker featurebase image with tag "$tag"

2
go.mod
View file

@ -73,6 +73,7 @@ require (
)
require (
github.com/google/uuid v1.3.0
github.com/jaffee/commandeer v0.5.0
github.com/linkedin/goavro/v2 v2.11.1
google.golang.org/grpc v1.46.0
@ -97,7 +98,6 @@ require (
github.com/golang/snappy v0.0.4 // indirect
github.com/google/btree v1.0.1 // indirect
github.com/google/pprof v0.0.0-20211214055906-6f57359322fd // indirect
github.com/google/uuid v1.3.0 // indirect
github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00 // indirect
github.com/gorilla/websocket v1.5.0 // indirect
github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 // indirect

19
idk/Makefile
View file

@ -1,6 +1,7 @@
GO ?= go
GOOS ?= $(shell $(GO) env GOOS)
GOARCH ?= $(shell $(GO) env GOARCH)
GOPATH ?= $(shell $(GO) env GOPATH)
GO_VERSION=1.17.8
GO_BUILD_FLAGS=
ODBC_ENABLED=
@ -309,3 +310,21 @@ docker-push-ecr: docker-image aws-login
aws-login:
aws sso login --profile $(PROFILE)
# Build mock implementations of AWS service API interfaces for unit testing.
#
# These are run-once/rarely targets as they produce code artifacts only used
# during `go test`. An automated CI/CD process does NOT need to invoke every time.
MOCKS = S3 Kinesis SQS
install-mock-generator:
@which $(GOPATH)/bin/mockery || (echo "Installing missing dependency 'mockery' to generate mock implementations for unit testing." && \
$(GO) install 'github.com/vektra/mockery/v2@latest')
update-mocks: $(addprefix update-mocks-, $(MOCKS))
update-mocks-%: install-mock-generator
$(eval AWS_SERVICE := $(shell echo $* | tr '[:upper:]' '[:lower:]'))
$(eval AWS_SDK_VERSION := $(shell grep 'github.com/aws/aws-sdk-go' ../go.mod | cut -d ' ' -f 2))
echo Generating mock for AWS service $(AWS_SERVICE) and SDK version $(AWS_SDK_VERSION) && \
$(GOPATH)/bin/mockery --name $*API --output idktest/mocks --filename $(AWS_SERVICE).go --dir $(GOPATH)/pkg/mod/github.com/aws/aws-sdk-go@$(AWS_SDK_VERSION)/service/$(AWS_SERVICE)/$(AWS_SERVICE)iface

32
idk/cmd/molecula-consumer-kinesis/main.go
View file

@ -9,24 +9,42 @@ import (
"github.com/molecula/featurebase/v3/logger"
)
func logFailure(errorType kinesis.ErrorType, m *kinesis.Main, v interface{}) {
log := m.Log()
if log == nil {
log = logger.NewStandardLogger(os.Stderr)
}
if errorType == kinesis.RecoverableErrorType {
log.Errorf("Error running command: %+v", v)
} else {
log.Panicf("Panic running command: %+v", v)
}
}
func main() {
m := kinesis.NewMain()
if err := pflag.LoadEnv(m, "CONSUMER_", nil); err != nil {
log.Fatal(err)
}
m.Rename()
// Capture any panic and log it before dying.
defer func() {
if r := recover(); r != nil {
logFailure(kinesis.PanicErrorType, m, r)
os.Exit(1)
}
}()
if m.DryRun {
log.Printf("%+v\n", m)
return
}
if err := m.Run(); err != nil {
log := m.Log()
if log == nil {
// if we fail before a logger was instantiated
logger.NewStandardLogger(os.Stderr).Errorf("Error running command: %v", err)
os.Exit(1)
}
log.Errorf("Error running command: %v", err)
logFailure(kinesis.RecoverableErrorType, m, err)
os.Exit(1)
}
}

19
idk/idktest/mocks/kinesis.go
View file

@ -1,9 +1,7 @@
// Code generated by mockery v1.0.0. DO NOT EDIT.
// Code generated by mockery v2.14.0. DO NOT EDIT.
package mocks
import "github.com/aws/aws-sdk-go/service/kinesis/kinesisiface"
import (
context "context"
@ -2456,4 +2454,17 @@ func (_m *KinesisAPI) WaitUntilStreamNotExistsWithContext(_a0 context.Context, _
return r0
}
var _ kinesisiface.KinesisAPI = (*KinesisAPI)(nil)
type mockConstructorTestingTNewKinesisAPI interface {
mock.TestingT
Cleanup(func())
}
// NewKinesisAPI creates a new instance of KinesisAPI. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
func NewKinesisAPI(t mockConstructorTestingTNewKinesisAPI) *KinesisAPI {
mock := &KinesisAPI{}
mock.Mock.Test(t)
t.Cleanup(func() { mock.AssertExpectations(t) })
return mock
}

19
idk/idktest/mocks/s3.go
View file

@ -1,9 +1,7 @@
// Code generated by mockery v1.0.0. DO NOT EDIT.
// Code generated by mockery v2.14.0. DO NOT EDIT.
package mocks
import "github.com/aws/aws-sdk-go/service/s3/s3iface"
import (
context "context"
@ -7821,4 +7819,17 @@ func (_m *S3API) WriteGetObjectResponseWithContext(_a0 context.Context, _a1 *s3.
return r0, r1
}
var _ s3iface.S3API = (*S3API)(nil)
type mockConstructorTestingTNewS3API interface {
mock.TestingT
Cleanup(func())
}
// NewS3API creates a new instance of S3API. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
func NewS3API(t mockConstructorTestingTNewS3API) *S3API {
mock := &S3API{}
mock.Mock.Test(t)
t.Cleanup(func() { mock.AssertExpectations(t) })
return mock
}

1662
idk/idktest/mocks/sqs.go Normal file
View file

File diff suppressed because it is too large Load diff

1
idk/ingest.go
View file

@ -191,6 +191,7 @@ func init() {
func (m *Main) PilosaClient() *pilosaclient.Client { return m.client }
func (m *Main) Log() logger.Logger { return m.log }
func (m *Main) SetLog(log logger.Logger) { m.log = log }
func NewMain() *Main {
fmt.Fprintf(os.Stderr, "Molecula Consumer %s, build time %s\n", Version, BuildTime)

40
idk/kinesis/cmd.go
View file

@ -3,10 +3,15 @@ package kinesis
import (
"time"
"github.com/molecula/featurebase/v3/idk"
"github.com/pkg/errors"
"github.com/molecula/featurebase/v3/idk"
)
// Main is the holder of all configurations for a Kinesis stream consumer.
//
// Along with the additional configuration fields, kinesis.Main also gains all
// fields and methods from idk.Main via composition.
type Main struct {
idk.Main `flag:"!embed"`
Timeout time.Duration `help:"Time to wait for more records from Kinesis before flushing a batch. 0 to disable."`
@ -16,8 +21,28 @@ type Main struct {
StreamName string `help:"Name of AWS Kinesis stream to consume records from."`
OffsetsPath string `help:"Path where the offsets file will be written. May be a path on the local filesystem, or an S3 URI."`
AWSProfile string `help:"Name of AWS profile to use. Alternatively, use environment variable AWS_PROFILE."`
ErrorQueueName string `help:"SQS queue name to send error and panic/runtime errors to."`
}
// NewMain returns a new instance of a Kinesis stream consumer configuration object.
//
// It specifies a callback NewSource that can be invoked to create a kinesis.Source object.
// This callback implicitly initializes an AWS session and uses that session to initialize
// clients to the following AWS resources: S3, Kinesis, and SQS. Client creation happens regardless
// of configuration. (ex: OffsetsPath and Header are local paths -> S3 client is created.)
//
// The default BatchSize is 20000 and Concurrency is 1. Any Concurrency value > 1 is NOT supported.
// These values are set on the returned kinesis.Main instance.
//
// The Logger instance on the kinesis.Source is always decorated when NewSource is invoked.
// Assuming no errors occur during AWS client initialization, the decorated Logger instance is
// propagated back to the kinesis.Main so that callers that configured it can also emit errors and
// panics to the SQS queue specified by ErrorQueueName. The behavior of the wrapped Logger depends on
// a non-empty ErrorQueueName, the existence of an SQS queue instance in AWSRegion with that name
// the StreamName field being of a particular format 'PREFIX'-VALID_UUID, and if a valid SQS queue URL
// can be resolved at the time of Logger initialization. If any of these are false, the error
// emission to an SQS queue functionality is not activated and the Logger instance behaves identically
// to its wrapped Logger and emits a warning to the caller that errors are not propagated to SQS.
func NewMain() *Main {
m := &Main{
Main: *idk.NewMain(),
@ -28,9 +53,9 @@ func NewMain() *Main {
m.OffsetMode = true
m.Main.Namespace = "ingester_kinesis"
m.Main.Pprof = "" // don't initialize pprof until we actually use it in tests
m.NewSource = func() (idk.Source, error) {
source := NewSource()
source.Log = m.Main.Log()
source.Timeout = m.Timeout
source.Header = m.Header
source.AWSRegion = m.AWSRegion
@ -39,10 +64,21 @@ func NewMain() *Main {
source.OffsetsPath = m.OffsetsPath
source.AWSProfile = m.AWSProfile
// This Logger instance is wrapped in `Open` -> `initAWS`.
source.Log = m.Main.Log()
source.ErrorQueueName = m.ErrorQueueName
err := source.Open()
if err != nil {
return nil, errors.Wrap(err, "opening source")
}
// `Open` succeeded -> AWS resources successfully initialized ->
// Logger instance was successfully wrapped. Now assign the wrapped
// Logger instance back to main so executables invoking this
// (ex: `molecula-consumer-kinesis`) will propagate errors and
// panics on failure using the wrapped Logger instance.
m.Main.SetLog(source.Log)
return source, nil
}
return m

234
idk/kinesis/logger.go Normal file
View file

@ -0,0 +1,234 @@
package kinesis
import (
"encoding/base64"
"encoding/json"
"fmt"
"strings"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/sqs"
"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
"github.com/google/uuid"
"github.com/pkg/errors"
"github.com/molecula/featurebase/v3/logger"
)
// ErrorType signifies the type of error encountered.
type ErrorType string
// These are the currently supported ErrorType values that can be emitted
// to an ErrorStore.
const (
RecoverableErrorType = ErrorType("Error")
PanicErrorType = ErrorType("Panic") // Runtime errors.
)
// ErrorStore is an abstraction over a resource like external storage, a database,
// a queue, etc. that can receive and store error messages.
type ErrorStore interface {
// Available checks if the backing resource can receive error
// messages via Push.
Available() bool
// Push emits an error message of type ErrorType to the backing resource.
//
// The caller should NOT assume that Available is called implicitly
// to check for ErrorStore availability.
Push(ErrorType, string, logger.Logger) error
}
// SinkErrorPayload contains all data about a single IDK error message that will be
// emitted to an ErrorStore; includes a timestamp and a valid sink ID.
//
// This payload is not meant to be used outside the context of error propogation to
// an ErrorStore.
type SinkErrorPayload struct {
SinkId string `json:"sink_id"`
ErrorType ErrorType `json:"error_type"`
ErrorMessage string `json:"error_msg"`
Timestamp string `json:"time"`
}
// SinkErrorQueue is an ErrorStore implementation that uses an SQS queue as its backing
// resource to emit error and panic messages to.
//
// It also maps 1-to-1 to a Kinesis stream via a unique sink ID.
type SinkErrorQueue struct {
sinkId string
name string
url string
queue sqsiface.SQSAPI
}
// NewSinkErrorQueue attempts to construct a SinkErrorQueue instance from an AWS SQS client,
// a queue name, and a sink ID.
//
// On success, callers can assume that a backing SQS queue resource exists and is fully initialized.
//
// Returns nil and an SQS error if an SQS queue URL cannot be resolved from the queue name and/or
// the AWS SQS client.
//
// This method assumes the sink ID argument is valid.
func NewSinkErrorQueue(queue sqsiface.SQSAPI, queueName, sinkId string) (*SinkErrorQueue, error) {
input := &sqs.GetQueueUrlInput{QueueName: &queueName}
output, err := queue.GetQueueUrl(input)
if err != nil {
return nil, err
}
return &SinkErrorQueue{sinkId, queueName, *output.QueueUrl, queue}, nil
}
// SinkErrorQueueFrom always constructs a SinkErrorQueue instance from an AWS SQS client and a
// kinesis.Source.
//
// Unlike NewSinkErrorQueue, this does NOT return an error if a queue URL cannot be resolved
// from the queue name and/or the AWS SQS client. Instead it will collapse to a SinkErrorQueue
// instance with a backing SQS resource that is ALWAYS unavailable. Attempting to invoke Push
// on this instance will not result in an error; instead it will just emit a warning that no
// backing SQS resource could be written to.
//
// This does check if the sink ID has a valid form: 'PREFIX'-VALID_UUID. If not, this collapses
// to a SinkErrorQueue instance that is ALWAYS unavailable.
func SinkErrorQueueFrom(queue sqsiface.SQSAPI, source *Source) *SinkErrorQueue {
// The below failure conditions are handled by collapsing to a no-op ErrorStore.Push implementation.
// - A missing SQS queue name.
// - An invalid sink ID (i.e. not a valid UUID); valid sink ID: "PREFIX"-UUID.
// - Unable to resolve queue URL from queue name.
//
// Means downstream ErrorStreamLogger behaves identical to its embedded Logger.
if source.ErrorQueueName == "" {
return &SinkErrorQueue{}
}
sinkId := strings.Join(strings.Split(source.StreamName, "-")[1:], "-")
_, err := uuid.Parse(sinkId)
if err != nil {
// Keep invalid sink ID around in case something downstream wants to log.
return &SinkErrorQueue{sinkId, source.ErrorQueueName, "", nil}
}
sinkErrorQueue, err := NewSinkErrorQueue(queue, source.ErrorQueueName, sinkId)
if err != nil {
return &SinkErrorQueue{sinkId, source.ErrorQueueName, "", nil}
}
return sinkErrorQueue
}
// Available checks that a valid SQS queue resource exists.
func (seq *SinkErrorQueue) Available() bool {
return seq.url != "" && seq.queue != nil
}
// Push attempts to emit a single error message of type ErrorType to a SQS queue resource.
//
// If the backing SQS queue resource is not available, this function is a no-op and does NOT
// return an error. Instead it emits a warning to the logger.Logger instance specified by the log
// argument.
//
// The warning can be ignored entirely by passing a nil log argument.
func (seq *SinkErrorQueue) Push(errorType ErrorType, message string, log logger.Logger) error {
if !seq.Available() {
if log != nil {
log.Warnf("Not pushing errors to an SQS queue='%+v' due to unavailability.", seq)
}
return nil
}
payload := SinkErrorPayload{
SinkId: seq.sinkId,
ErrorType: errorType,
ErrorMessage: message,
Timestamp: time.Now().Format(time.RFC3339),
}
payloadBytes, err := json.Marshal(&payload)
if err != nil {
msgTemplate := "Unable to marshal payload='%+v' to send message='%s' to queue='%+v'."
return errors.Wrap(err, fmt.Sprintf(msgTemplate, payload, message, seq.queue))
}
encodedPayload := base64.URLEncoding.EncodeToString(payloadBytes)
input := &sqs.SendMessageInput{
DelaySeconds: aws.Int64(10),
MessageBody: aws.String(encodedPayload),
QueueUrl: &seq.url,
}
_, err = seq.queue.SendMessage(input)
if err != nil {
msgTemplate := "Unable to send message='%s' to queue='%+v' with input='%+v'."
return errors.Wrap(err, fmt.Sprintf(msgTemplate, message, seq.queue, input))
}
return nil
}
// ErrorStreamLogger is a logger.Logger implementation that decorates a base logger.Logger instance
// and emits error and panic messages to an ErrorStore.
//
// All other log levels delegate to the base logger.Logger implementation.
type ErrorStreamLogger struct {
base logger.Logger
store ErrorStore
}
// NewErrorStreamLogger constructs an ErrorStreamLogger from a logger.Logger and an ErrorStore.
func NewErrorStreamLogger(base logger.Logger, store ErrorStore) *ErrorStreamLogger {
return &ErrorStreamLogger{base, store}
}
// Printf just delegates to the wrapped/base logger.Logger's Printf implementation.
func (esl *ErrorStreamLogger) Printf(format string, v ...interface{}) {
esl.base.Printf(format, v...)
}
// Debugf just delegates to the wrapped/base logger.Logger's Debugf implementation.
func (esl *ErrorStreamLogger) Debugf(format string, v ...interface{}) {
esl.base.Debugf(format, v...)
}
// Infof just delegates to the wrapped/base logger.Logger's Infof implementation.
func (esl *ErrorStreamLogger) Infof(format string, v ...interface{}) {
esl.base.Infof(format, v...)
}
// Warnf just delegates to the wrapped/base logger.Logger's Warnf implementation.
func (esl *ErrorStreamLogger) Warnf(format string, v ...interface{}) {
esl.base.Warnf(format, v...)
}
// Errorf delegates to the wrapped/base logger.Logger's Errorf implementation and additionally/
// pushes an error message with RecoverableErrorType ErrorType to the ErrorStore.
//
// If an error occurs during a Push to the ErrorStore, the error is logged to the wrapped/base
// logger.Logger using Errorf.
func (esl *ErrorStreamLogger) Errorf(format string, v ...interface{}) {
esl.base.Errorf(format, v...)
err := esl.store.Push(RecoverableErrorType, fmt.Sprintf(format, v...), esl)
if err != nil {
errMsg := fmt.Sprintf("Failed during push to store='%+v'", esl.store)
esl.base.Errorf(errors.Wrap(err, errMsg).Error())
}
}
// Panicf delegates to the wrapped/base logger.Logger's Panicf implementation and additionally
// pushes an error message with PanicErrorType ErrorType to the ErrorStore.
//
// If an error occurs during a Push to the ErrorStore, the error is logged to the wrapped/base
// logger.Logger using Errorf NOT Panicf.
func (esl *ErrorStreamLogger) Panicf(format string, v ...interface{}) {
esl.base.Panicf(format, v...)
err := esl.store.Push(PanicErrorType, fmt.Sprintf(format, v...), esl)
if err != nil {
// A (recoverable) error occurred during a push to the store, so
// log that error using `Errorf` not `Panicf`.
errMsg := fmt.Sprintf("Failed during push to store='%+v'", esl.store)
esl.base.Errorf(errors.Wrap(err, errMsg).Error())
}
}

540
idk/kinesis/logger_test.go Normal file
View file

@ -0,0 +1,540 @@
package kinesis
import (
"encoding/base64"
"encoding/json"
"fmt"
"strings"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/sqs"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/molecula/featurebase/v3/idk/idktest/mocks"
"github.com/molecula/featurebase/v3/logger"
)
// In-memory featurebase/logger.Logger implementation that stores by level
// {"PRINTF", "DEBUG", "INFO", "WARN", "ERROR", "PANIC"} useful for unit tests.
type MapStashLogger struct {
messages map[string][]string
}
func NewMapStashLogger() *MapStashLogger {
messages := make(map[string][]string)
messages["PRINTF"] = []string{}
messages["DEBUG"] = []string{}
messages["INFO"] = []string{}
messages["WARN"] = []string{}
messages["ERROR"] = []string{}
messages["PANIC"] = []string{}
return &MapStashLogger{messages}
}
func (msl *MapStashLogger) Printf(format string, v ...interface{}) {
msl.messages["PRINTF"] = append(msl.messages["PRINTF"], fmt.Sprintf(format, v...))
}
func (msl *MapStashLogger) Debugf(format string, v ...interface{}) {
msl.messages["DEBUG"] = append(msl.messages["DEBUG"], fmt.Sprintf(format, v...))
}
func (msl *MapStashLogger) Infof(format string, v ...interface{}) {
msl.messages["INFO"] = append(msl.messages["INFO"], fmt.Sprintf(format, v...))
}
func (msl *MapStashLogger) Warnf(format string, v ...interface{}) {
msl.messages["WARN"] = append(msl.messages["WARN"], fmt.Sprintf(format, v...))
}
func (msl *MapStashLogger) Errorf(format string, v ...interface{}) {
msl.messages["ERROR"] = append(msl.messages["ERROR"], fmt.Sprintf(format, v...))
}
func (msl *MapStashLogger) Panicf(format string, v ...interface{}) {
msl.messages["PANIC"] = append(msl.messages["PANIC"], fmt.Sprintf(format, v...))
}
// In-memory error store implementation to use in unit tests.
type MapStashErrorStore struct {
storage map[ErrorType][]string
available bool
simulatedError error
}
func NewMapStashErrorStore(available bool, err error) *MapStashErrorStore {
return &MapStashErrorStore{make(map[ErrorType][]string), available, err}
}
func (mses *MapStashErrorStore) Available() bool {
return mses.available
}
func (mses *MapStashErrorStore) Push(errorType ErrorType, message string, log logger.Logger) error {
if !mses.Available() {
return nil
}
if mses.simulatedError != nil {
return mses.simulatedError
}
mses.storage[errorType] = append(mses.storage[errorType], message)
return nil
}
func TestNewSinkErrorQueueSuccess(t *testing.T) {
mockSQS := &mocks.SQSAPI{}
mockGetQueueUrlFn := func(input *sqs.GetQueueUrlInput) *sqs.GetQueueUrlOutput {
name := *input.QueueName
url := fmt.Sprintf("https://unit-test.queue.%s.url", name)
return &sqs.GetQueueUrlOutput{QueueUrl: aws.String(url)}
}
mockSQS.On("GetQueueUrl", mock.MatchedBy(func(input *sqs.GetQueueUrlInput) bool {
return input.QueueName != nil && *input.QueueName != ""
})).Return(mockGetQueueUrlFn, nil)
queue, err := NewSinkErrorQueue(mockSQS, "dummy-123", "a-b-c")
assert.Nil(t, err)
assert.Equal(t, "a-b-c", queue.sinkId)
assert.Equal(t, "dummy-123", queue.name)
assert.Equal(t, "https://unit-test.queue.dummy-123.url", queue.url)
assert.Same(t, mockSQS, queue.queue)
}
func TestNewSinkErrorQueueFailSQSGetQueueUrlErrored(t *testing.T) {
mockSQS := &mocks.SQSAPI{}
mockGetQueueUrlFn := func(input *sqs.GetQueueUrlInput) *sqs.GetQueueUrlOutput {
// Value ignored by the caller on failure.
return &sqs.GetQueueUrlOutput{}
}
errMsg := "Could not retrieve queue URL."
mockSQS.On("GetQueueUrl", mock.MatchedBy(func(input *sqs.GetQueueUrlInput) bool {
return true
})).Return(mockGetQueueUrlFn, errors.New(errMsg))
queue, err := NewSinkErrorQueue(mockSQS, "dummy-123", "a-b-c")
assert.Nil(t, queue)
assert.NotNil(t, err)
assert.Equal(t, errMsg, err.Error())
}
func TestSinkErrorQueueFromSuccess(t *testing.T) {
mockSQS := &mocks.SQSAPI{}
mockGetQueueUrlFn := func(input *sqs.GetQueueUrlInput) *sqs.GetQueueUrlOutput {
name := *input.QueueName
url := fmt.Sprintf("https://unit-test.queue.%s.url", name)
return &sqs.GetQueueUrlOutput{QueueUrl: aws.String(url)}
}
mockSQS.On("GetQueueUrl", mock.MatchedBy(func(input *sqs.GetQueueUrlInput) bool {
return input.QueueName != nil && *input.QueueName != ""
})).Return(mockGetQueueUrlFn, nil)
validUuid := "0f2af1d9-52db-4a1c-bc75-554d84b01850"
streamName := fmt.Sprintf("sink-%s", validUuid) // Structure mimics how cloud ECS instance names Kinesis stream.
source := &Source{
ErrorQueueName: "my-queue-01234",
StreamName: streamName,
}
var queue *SinkErrorQueue
queue = SinkErrorQueueFrom(mockSQS, source)
assert.Equal(t, validUuid, queue.sinkId)
assert.Equal(t, "my-queue-01234", queue.name)
assert.Equal(t, "https://unit-test.queue.my-queue-01234.url", queue.url)
assert.Same(t, mockSQS, queue.queue)
}
func TestSinkErrorQueueFromFailMissingQueueName(t *testing.T) {
// Missing queue name on `Source` object creates a failure condition before SQS instance is even inspected.
mockSQS := &mocks.SQSAPI{}
validUuid := "0f2af1d9-52db-4a1c-bc75-554d84b01850"
streamName := fmt.Sprintf("sink-%s", validUuid) // Structure mimics how cloud ECS instance names its Kinesis stream.
source := &Source{StreamName: streamName}
var queue *SinkErrorQueue
queue = SinkErrorQueueFrom(mockSQS, source)
assert.Equal(t, "", queue.sinkId)
assert.Equal(t, "", queue.name)
assert.Equal(t, "", queue.url)
assert.Nil(t, queue.queue)
}
func TestSinkErrorQueueFromFailMalformedSinkId(t *testing.T) {
mockSQS := &mocks.SQSAPI{}
mockGetQueueUrlFn := func(input *sqs.GetQueueUrlInput) *sqs.GetQueueUrlOutput {
// Value ignored by the caller on failure.
return &sqs.GetQueueUrlOutput{}
}
mockSQS.On("GetQueueUrl", mock.MatchedBy(func(input *sqs.GetQueueUrlInput) bool {
return true
})).Return(mockGetQueueUrlFn, nil)
malformedUuid := "28hsdfas636bd"
streamName := fmt.Sprintf("sink-%s", malformedUuid)
source := &Source{
ErrorQueueName: "valid-queue-90123",
StreamName: streamName,
}
var queue *SinkErrorQueue
queue = SinkErrorQueueFrom(mockSQS, source)
assert.Equal(t, malformedUuid, queue.sinkId) // Keeps invalid sink ID around for downstream logging purposes.
assert.Equal(t, "valid-queue-90123", queue.name)
assert.Equal(t, "", queue.url) // Did not reach a point where `sqs.GetQueueUrl` is even called.
assert.Nil(t, queue.queue)
}
func TestSinkErrorQueueFromFailEmptyStreamNameDoesNotCausePanic(t *testing.T) {
mockSQS := &mocks.SQSAPI{}
mockGetQueueUrlFn := func(input *sqs.GetQueueUrlInput) *sqs.GetQueueUrlOutput {
// Value ignored by the caller on failure.
return &sqs.GetQueueUrlOutput{}
}
mockSQS.On("GetQueueUrl", mock.MatchedBy(func(input *sqs.GetQueueUrlInput) bool {
return true
})).Return(mockGetQueueUrlFn, nil)
source := &Source{
ErrorQueueName: "valid-queue-88888888888",
StreamName: "",
}
var queue *SinkErrorQueue
queue = SinkErrorQueueFrom(mockSQS, source)
assert.Equal(t, "", queue.sinkId) // Keeps invalid sink ID around for downstream logging purposes.
assert.Equal(t, "valid-queue-88888888888", queue.name)
assert.Equal(t, "", queue.url) // Did not reach a point where `sqs.GetQueueUrl` is even called.
assert.Nil(t, queue.queue)
}
func TestSinkErrorQueueFromFailSQSGetQueueUrlErrored(t *testing.T) {
mockSQS := &mocks.SQSAPI{}
mockGetQueueUrlFn := func(input *sqs.GetQueueUrlInput) *sqs.GetQueueUrlOutput {
// Value ignored by the caller on failure.
return &sqs.GetQueueUrlOutput{}
}
errMsg := "This error message isn't propagated but is handled within the method."
mockSQS.On("GetQueueUrl", mock.MatchedBy(func(input *sqs.GetQueueUrlInput) bool {
return true
})).Return(mockGetQueueUrlFn, errors.New(errMsg))
validUuid := "fa6f1631-7c29-4023-95ac-48a2f59fae9b"
streamName := fmt.Sprintf("sink-%s", validUuid) // Structure mimics how cloud ECS instance names Kinesis stream.
source := &Source{
ErrorQueueName: "queue-that-we-stole-962463",
StreamName: streamName,
}
var queue *SinkErrorQueue
queue = SinkErrorQueueFrom(mockSQS, source)
assert.Equal(t, validUuid, queue.sinkId) // Keeps valid sink ID around for downstream logging purposes.
assert.Equal(t, "queue-that-we-stole-962463", queue.name)
assert.Equal(t, "", queue.url) // Could not resolve queue URL, so this remains empty.
assert.Nil(t, queue.queue)
}
func TestSinkErrorQueueAvailable(t *testing.T) {
emptySeq := &SinkErrorQueue{}
assert.False(t, emptySeq.Available())
// Only depends on URL being non-empty and a non-nil reference to an SQS queue.
mockSQS := &mocks.SQSAPI{}
seq := &SinkErrorQueue{
url: "asdfadfasdf2332798",
queue: mockSQS,
}
assert.True(t, seq.Available())
}
func TestSinkErrorQueuePushWhenQueueNotAvailableDoesNotThrowErrorAndNoOps(t *testing.T) {
emptySeq := &SinkErrorQueue{}
// Logger is nil.
err := emptySeq.Push(RecoverableErrorType, "werqw zxcw7228323974", nil)
assert.Nil(t, err)
// Logger is not nil; check that a warning is issued using the Logger instance.
logger := NewMapStashLogger()
err2 := emptySeq.Push(RecoverableErrorType, "werqw zxcw7228323974", logger)
assert.Nil(t, err2)
assert.Equal(t, 1, len(logger.messages["WARN"]))
// Check warning is emitted to the logger.
assert.True(t, strings.HasPrefix(logger.messages["WARN"][0], "Not pushing errors to an SQS queue="))
}
func TestSinkErrorQueuePushSuccess(t *testing.T) {
mockSQS := &mocks.SQSAPI{}
// Use variables defined outside the scope of a closure to save the message body and queueURL.
var actualBody string
var actualQueueUrl string
mockSendMessageFn := func(input *sqs.SendMessageInput) *sqs.SendMessageOutput {
actualBody = *input.MessageBody
actualQueueUrl = *input.QueueUrl
return &sqs.SendMessageOutput{MessageId: aws.String(actualBody[:5])}
}
mockSQS.On("SendMessage", mock.MatchedBy(func(input *sqs.SendMessageInput) bool {
return input.QueueUrl != nil && *input.QueueUrl != ""
})).Return(mockSendMessageFn, nil)
seq := &SinkErrorQueue{
sinkId: "xxx-000",
name: "pushover",
url: "https://unit-test.queue.push.url",
queue: mockSQS,
}
errMsg := "079 cxmn, 198sfakjl"
err := seq.Push(PanicErrorType, errMsg, nil)
assert.Nil(t, err)
// Verify queue URL.
assert.Equal(t, seq.url, actualQueueUrl)
// Now check message body.
decoded, err := base64.URLEncoding.DecodeString(actualBody)
assert.Nil(t, err, "Error while base64 URL decoding.")
payload := SinkErrorPayload{}
err = json.Unmarshal(decoded, &payload)
assert.Nil(t, err, "Error while de-serializing JSON into SinkErrorPayload object.") // An error de-serializing JSON fails the unit test.
assert.Equal(t, "xxx-000", payload.SinkId)
assert.Equal(t, "Panic", string(payload.ErrorType))
assert.Equal(t, errMsg, payload.ErrorMessage)
assert.GreaterOrEqual(t, time.Now().Format(time.RFC3339), payload.Timestamp) // Now's timestamp should be later than timestamp of sent message.
}
func TestNewErrorStreamLogger(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(true, nil)
logger := NewErrorStreamLogger(baseLogger, store)
assert.Same(t, baseLogger, logger.base)
assert.Same(t, store, logger.store)
}
func TestErrorStreamLoggerPrintf(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(true, nil)
logger := NewErrorStreamLogger(baseLogger, store)
logger.Printf("printf:%d", 3)
assert.Equal(t, 1, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 0, len(baseLogger.messages["ERROR"]))
assert.Equal(t, 0, len(baseLogger.messages["PANIC"]))
assert.Equal(t, "printf:3", baseLogger.messages["PRINTF"][0])
assert.Equal(t, 0, len(store.storage[RecoverableErrorType]))
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
}
func TestErrorStreamLoggerDebugf(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(true, nil)
logger := NewErrorStreamLogger(baseLogger, store)
logger.Debugf("debug::%d", 4)
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 1, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 0, len(baseLogger.messages["ERROR"]))
assert.Equal(t, 0, len(baseLogger.messages["PANIC"]))
assert.Equal(t, "debug::4", baseLogger.messages["DEBUG"][0])
assert.Equal(t, 0, len(store.storage[RecoverableErrorType]))
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
}
func TestErrorStreamLoggerInfof(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(true, nil)
logger := NewErrorStreamLogger(baseLogger, store)
logger.Infof("info:::%d", 5)
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 1, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 0, len(baseLogger.messages["ERROR"]))
assert.Equal(t, 0, len(baseLogger.messages["PANIC"]))
assert.Equal(t, "info:::5", baseLogger.messages["INFO"][0])
assert.Equal(t, 0, len(store.storage[RecoverableErrorType]))
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
}
func TestErrorStreamLoggerWarnf(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(true, nil)
logger := NewErrorStreamLogger(baseLogger, store)
logger.Warnf("warn::::%d", 6)
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 1, len(baseLogger.messages["WARN"]))
assert.Equal(t, 0, len(baseLogger.messages["ERROR"]))
assert.Equal(t, 0, len(baseLogger.messages["PANIC"]))
assert.Equal(t, "warn::::6", baseLogger.messages["WARN"][0])
assert.Equal(t, 0, len(store.storage[RecoverableErrorType]))
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
}
func TestErrorStreamLoggerErrorfWithAvailableQueue(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(true, nil) // Queue is available so a push actually occurs.
logger := NewErrorStreamLogger(baseLogger, store)
logger.Errorf("error:::::%d", 7)
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 1, len(baseLogger.messages["ERROR"]))
assert.Equal(t, 0, len(baseLogger.messages["PANIC"]))
assert.Equal(t, "error:::::7", baseLogger.messages["ERROR"][0])
assert.Equal(t, 1, len(store.storage[RecoverableErrorType])) // Verifies a push occurred.
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
assert.Equal(t, "error:::::7", store.storage[RecoverableErrorType][0])
}
func TestErrorStreamLoggerErrorfNoAvailableQueue(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(false, nil) // Queue is unavailable.
logger := NewErrorStreamLogger(baseLogger, store)
logger.Errorf("error:::::%d%s", 7, "a")
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 1, len(baseLogger.messages["ERROR"])) // Verifies base logger is delegated to even when queue is unavailable.
assert.Equal(t, 0, len(baseLogger.messages["PANIC"]))
assert.Equal(t, "error:::::7a", baseLogger.messages["ERROR"][0])
assert.Equal(t, 0, len(store.storage[RecoverableErrorType]))
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
}
func TestErrorStreamLoggerErrorfPushFailEmitsErrorToBaseLogger(t *testing.T) {
baseLogger := NewMapStashLogger()
errMsg := "ERROR: this error should be emitted to the base logger."
store := NewMapStashErrorStore(true, errors.New(errMsg)) // Simulate error condition during push.
logger := NewErrorStreamLogger(baseLogger, store)
logger.Errorf("error:::::%d%s", 7, "aa")
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 2, len(baseLogger.messages["ERROR"])) // Base logger should have a record of both errors and emit both.
assert.Equal(t, 0, len(baseLogger.messages["PANIC"]))
assert.Equal(t, "error:::::7aa", baseLogger.messages["ERROR"][0])
assert.True(t, strings.Contains(baseLogger.messages["ERROR"][1], errMsg)) // Check original push error message wrapped.
assert.True(t, strings.Contains(baseLogger.messages["ERROR"][1], "Failed during push to store=")) // Check error contains info that it occurred during a push.
assert.Equal(t, 0, len(store.storage[RecoverableErrorType]))
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
}
func TestErrorStreamLoggerPanicfWithAvailableQueue(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(true, nil) // Queue is available so a push actually occurs.
logger := NewErrorStreamLogger(baseLogger, store)
logger.Panicf("panic::::::%d", 8)
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 0, len(baseLogger.messages["ERROR"]))
assert.Equal(t, 1, len(baseLogger.messages["PANIC"]))
assert.Equal(t, "panic::::::8", baseLogger.messages["PANIC"][0])
assert.Equal(t, 0, len(store.storage[RecoverableErrorType])) // Verifies a push occurred.
assert.Equal(t, 1, len(store.storage[PanicErrorType]))
assert.Equal(t, "panic::::::8", store.storage[PanicErrorType][0])
}
func TestErrorStreamLoggerPanicfNoAvailableQueue(t *testing.T) {
baseLogger := NewMapStashLogger()
store := NewMapStashErrorStore(false, nil) // Queue is unavailable.
logger := NewErrorStreamLogger(baseLogger, store)
logger.Panicf("panic::::::%d%s", 8, "b")
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 0, len(baseLogger.messages["ERROR"]))
assert.Equal(t, 1, len(baseLogger.messages["PANIC"])) // Verifies base logger is delegated to even when queue is unavailable.
assert.Equal(t, "panic::::::8b", baseLogger.messages["PANIC"][0])
assert.Equal(t, 0, len(store.storage[RecoverableErrorType]))
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
}
func TestErrorStreamLoggerPanicfPushFailEmitsErrorToBaseLogger(t *testing.T) {
baseLogger := NewMapStashLogger()
errMsg := "PANIC: this error should be emitted to the base logger."
store := NewMapStashErrorStore(true, errors.New(errMsg)) // Simulate error condition during push.
logger := NewErrorStreamLogger(baseLogger, store)
logger.Panicf("panic::::::%d%s", 8, "bb")
assert.Equal(t, 0, len(baseLogger.messages["PRINTF"]))
assert.Equal(t, 0, len(baseLogger.messages["DEBUG"]))
assert.Equal(t, 0, len(baseLogger.messages["INFO"]))
assert.Equal(t, 0, len(baseLogger.messages["WARN"]))
assert.Equal(t, 1, len(baseLogger.messages["ERROR"])) // Base logger should have a record of the error due to a push.
assert.Equal(t, 1, len(baseLogger.messages["PANIC"])) // Base logger should have a record of the original panic/runtime error.
assert.Equal(t, "panic::::::8bb", baseLogger.messages["PANIC"][0])
assert.True(t, strings.Contains(baseLogger.messages["ERROR"][0], errMsg)) // Check original push error message wrapped.
assert.True(t, strings.Contains(baseLogger.messages["ERROR"][0], "Failed during push to store=")) // Check error contains info that it occurred during a push.
assert.Equal(t, 0, len(store.storage[RecoverableErrorType]))
assert.Equal(t, 0, len(store.storage[PanicErrorType]))
}

9
idk/kinesis/source.go
View file

@ -15,6 +15,7 @@ import (
"github.com/aws/aws-sdk-go/service/kinesis/kinesisiface"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/aws/aws-sdk-go/service/s3/s3iface"
"github.com/aws/aws-sdk-go/service/sqs"
"github.com/molecula/featurebase/v3/idk"
"github.com/molecula/featurebase/v3/idk/internal"
@ -37,6 +38,8 @@ type Source struct {
StreamName string
OffsetsPath string
ErrorQueueName string
schema []idk.Field
paths idk.PathTable
@ -172,11 +175,17 @@ func (s *Source) initAWS() error {
s.Log.Infof("Overriding default AWS region: %s", s.AWSRegion)
config.Region = aws.String(s.AWSRegion)
}
sess, err := session.NewSession(config)
if err != nil {
return errors.Wrap(err, "creating AWS session")
}
s.session = sess
// Wrap the Logger instance to additionally broadcast errors and panics into an SQS queue.
queue := sqs.New(sess)
s.Log = NewErrorStreamLogger(s.Log, SinkErrorQueueFrom(queue, s))
s.s3client = s3.New(sess)
s.kinesisClient = kinesis.New(sess)
return nil