Add Kinesis support by marceloboeira · Pull Request #151 · openflagr/flagr

marceloboeira · 2018-08-04T11:31:00Z

From: #150

Description

Add Kinesis support for the data recorder.

Motivation and Context

That makes it more flexible since Kafka is not easy to self-host and Kinesis works well-enough for small throughput rates.

How Has This Been Tested?

Still figuring how we should test this since Kinesis is not open source, mocking? or just best effort (unit + offline integration?)

Types of changes

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to change)

Checklist:

My code follows the code style of this project.
My change requires a change to the documentation.
I have updated the documentation accordingly.
I have added tests to cover my changes.
All new and existing tests passed.
Delivery Test

Delivery Test

For the delivery test I've created a small script to trigger 10k requests, to the evaluation endpoint, in parallel (batches of 8/10). That should put at least 10k records into the stream.

e.g.:

seq 1 10000 | parallel -I {} http --ignore-stdin post 'http://127.0.0.1:18000/api/v1/evaluation'  entityID=neu-{} entityType=report flagID:=3 > neu.txt

In the meantime I was reading from the stream and storing the content into a file. Later on I've grep the content of the file, looking for the IDs of the entities sent, filtering that out, sorting, checking if its uniq and counting the uniq IDs.

grep "neu-[0-9]*" -ao kinesis.log | sort -V | uniq -c | sort

That returned 10k.

codecov-io · 2018-08-04T11:34:22Z

Codecov Report

Merging #151 into master will decrease coverage by 0.37%.
The diff coverage is 83.67%.

@@            Coverage Diff             @@
##           master     #151      +/-   ##
==========================================
- Coverage   93.54%   93.16%   -0.38%     
==========================================
  Files          23       24       +1     
  Lines        1193     1244      +51     
==========================================
+ Hits         1116     1159      +43     
- Misses         54       58       +4     
- Partials       23       27       +4

Impacted Files	Coverage Δ
pkg/handler/data_recorder.go	`100% <100%> (ø)`	⬆️
pkg/handler/data_recorder_kinesis.go	`82.97% <82.97%> (ø)`

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update b942c88...0a2f262. Read the comment docs.

zhouzhuojie · 2018-08-06T18:17:10Z

+func TestGetDataRecorderWhenKinesis(t *testing.T) {
+	singletonDataRecorderOnce = sync.Once{}
+	defer gostub.StubFunc(&NewKinesisRecorder, nil).Reset()
+	config.Config.RecorderType = "kinesis"


Let's also set the RecorderType back to default after this test

zhouzhuojie · 2018-08-06T18:20:17Z

+	*models.EvalResult
+}
+
+// Unmalshalls the context


nit, comment looks wrong. Do you mean Payload marshals the EvalResult?

zhouzhuojie · 2018-08-06T18:33:13Z

+		logrus.WithField("kinesis_error", err).Error("error marshaling")
+	}
+
+	err = k.producer.Put(payload, kr.Key())


Currently the kafka record schema looks like

{ "payload": string(EvalResult.MarshalBinary()) OR encrypted string, "encrypted": false <---- optional, you can ignore this and implement encryption later if needed }

Can we make the same abstraction/schema in kinesis? So that it's seamless to switch data recorders.

Another idea is to move https://github.com/checkr/flagr/blob/master/pkg/handler/data_recorder_kafka.go#L152 as the shared "message frame" for both Kafka and Kinesis

I saw a lot of similarities, thought of doing that right away, but it could get messy thinking of versioning and commit history. What about starting with the duplication, making everything isolated and working, and then I can open a second PR with the abstractions, so first I can implement and figure what will be duplicated. WDYT?

Sounds good for deferring the abstractions later. Let's at least make sure the message schema is the same, otherwise, there will be breaking changes for Kinesis in the future.

zhouzhuojie · 2018-08-06T18:35:08Z

+
+	go func() {
+		for err := range p.NotifyFailures() {
+			logrus.WithField("kinesis_error", err).Error("error pushing")


nit, the error message can be more specific, like logrus.WithField("kinesis_error", err).Error("error pushing to kinesis"), same for the error messages below

zhouzhuojie · 2018-08-06T18:35:35Z

+
+	"github.com/checkr/flagr/pkg/util"
+	"github.com/checkr/flagr/swagger_gen/models"
+	//"github.com/prashantv/gostub"


remove comments

zhouzhuojie · 2018-08-06T18:36:17Z

 	RecorderKafkaEncryptionKey  string        `env:"FLAGR_RECORDER_KAFKA_ENCRYPTION_KEY" envDefault:""`

+	// Kinesis related configurations for data records logging (Flagr Metrics)
+	RecorderKinesisStreamName          string        `env:"FLAGR_RECORDER_KINESIS_STREAM_NAME" envDefault:"flagr-stream"`


can be the same default as kafka, i.e. flagr-records

zhouzhuojie · 2018-08-06T18:37:24Z

+	RecorderKinesisAggregateBatchCount int           `env:"FLAGR_RECORDER_KINESIS_AGGREGATE_BATCH_COUNT" envDefault:"4294967295"`
+	RecorderKinesisAggregateBatchSize  int           `env:"FLAGR_RECORDER_KINESIS_AGGREGATE_BATCH_SIZE" envDefault:"51200"`
+	RecorderKinesisVerbose             bool          `env:"FLAGR_RECORDER_KINESIS_VERBOSE" envDefault:"false"`
+


Also, what about kinesis connection credentials, not very familiar with kinesis, I would image you may need AWS keys.

Yes, you need, but they are loaded implicitly by default from AWS packages, we can document but I wouldn't change the that behaviour.

👍 it would be great if we can have some docs that point to the AWS behavior.

zhouzhuojie

Thanks for the PR! LGTM. It would be great if we can cover more lines.

zhouzhuojie · 2018-08-08T18:39:16Z

+}
+
+// NewKinesisRecorder creates a new Kinesis recorder
+var NewKinesisRecorder = func() DataRecorder {


it would be great if we can cover more lines

https://codecov.io/gh/checkr/flagr/pull/151/diff?src=pr&el=tree#diff-cGtnL2hhbmRsZXIvZGF0YV9yZWNvcmRlcl9raW5lc2lzLmdv

Yes, it's still WIP, I have to test more things. I just didn't have time yet.

marceloboeira · 2018-08-10T11:16:10Z

@zhouzhuojie I think the code its pretty much done, please take a look. I'll be conducting some performance/delivery tests now to make sure like if I call 1M times, I have 1M records on the stream...

Once I have that ready I'll mention here so we move forward. In the meantime, please feel free to look at the code and I can work on your comments right away.

marceloboeira · 2018-08-10T13:26:58Z

Delivery Test

For the delivery test I've created a small script to trigger 10k requests, to the evaluation endpoint, in parallel (batches of 8/10). That should put at least 10k records into the stream.

e.g.:

seq 1 10000 | parallel -I {} http --ignore-stdin post 'http://127.0.0.1:18000/api/v1/evaluation'  entityID=neu-{} entityType=report flagID:=3 > neu.txt

In the meantime I was reading from the stream and storing the content into a file. Later on I've grep the content of the file, looking for the IDs of the entities sent, filtering that out, sorting, checking if its uniq and counting the uniq IDs.

grep "neu-[0-9]*" -ao kinesis.log | sort -V | uniq -c | sort

That returned 10k.

openflagr#150 Environment defaults: https://github.com/a8m/kinesis-producer/blob/master/config.go#L12 - Add Kinesis data recorder (without encryption) - Add documentation for AWS Authentication with Kinesis - Increase test coverage of the data recorder Delivery test The delivery tests ran with a small script to trigger 10k requests, to the evaluation endpoint, in parallel (batches of 8/10). That should `put` at least 10k records into the stream. e.g.: ```script seq 1 10000 | parallel -I {} http --ignore-stdin post 'http://127.0.0.1:18000/api/v1/evaluation' entityID=neu-{} entityType=report flagID:=3 > neu.txt ``` While reading from the stream and storing the content into a file. By greping the content of the file, looking for the IDs of the entities sent, filtering that out, sorting, checking if its uniq and counting the uniq IDs. `grep "neu-[0-9]*" -ao kinesis.log | sort -V | uniq -c | sort` That returned 10k.

zhouzhuojie

🎉

marceloboeira force-pushed the feature/kinesis-data-recorder branch 4 times, most recently from 151c539 to 34fdfc8 Compare August 6, 2018 12:17

zhouzhuojie reviewed Aug 6, 2018

View reviewed changes

marceloboeira force-pushed the feature/kinesis-data-recorder branch 3 times, most recently from 8615334 to 70d2ef1 Compare August 8, 2018 09:47

zhouzhuojie reviewed Aug 8, 2018

View reviewed changes

marceloboeira force-pushed the feature/kinesis-data-recorder branch 2 times, most recently from 740d49a to babad0c Compare August 10, 2018 09:36

marceloboeira changed the title ~~[WIP] Add Kinesis support~~ Add Kinesis support Aug 10, 2018

marceloboeira force-pushed the feature/kinesis-data-recorder branch from babad0c to 0a2f262 Compare August 10, 2018 13:30

zhouzhuojie approved these changes Aug 10, 2018

View reviewed changes

zhouzhuojie merged commit 3e8caa2 into openflagr:master Aug 10, 2018

marceloboeira mentioned this pull request Aug 11, 2018

Kinesis Data Recorder Adapter #150

Closed

marceloboeira deleted the feature/kinesis-data-recorder branch August 17, 2018 08:49

Conversation

marceloboeira commented Aug 4, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Motivation and Context

How Has This Been Tested?

Types of changes

Checklist:

Delivery Test

Uh oh!

codecov-io commented Aug 4, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

marceloboeira Aug 7, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

marceloboeira Aug 7, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhouzhuojie left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

marceloboeira commented Aug 10, 2018

Uh oh!

marceloboeira commented Aug 10, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Delivery Test

Uh oh!

zhouzhuojie left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

marceloboeira commented Aug 4, 2018 •

edited

Loading

codecov-io commented Aug 4, 2018 •

edited

Loading

marceloboeira Aug 7, 2018 •

edited

Loading

marceloboeira Aug 7, 2018 •

edited

Loading

marceloboeira commented Aug 10, 2018 •

edited

Loading