CUMULUS-359 Add payload to one time rule created from kinesis stream #232

laurenfrederick · 2018-03-05T20:37:19Z

Summary: Added payload to the rule created for the kinesis message to pass the message through to the workflow

Addresses CUMULUS-359: Kinesis triggered workflows do not retain any of the triggering message's body

Changes

Added payload to rule
Added to the test
Updated code and test file for ESLint

Test Plan

Things that should succeed before merging.

Unit tests
Adhoc testing

Reviewers: @abarciauskas-bgse @jennyhliu

laurenfrederick

Will update Changelog after current release is published

scisco · 2018-03-05T21:54:35Z

packages/api/lambdas/kinesis-consumer.js

 */
-async function createOneTimeRules(kinesisRules) {
+async function createOneTimeRules(kinesisRules, eventObject) {


This is not scalable at its current form. We should update this function to only generate the cumulus message and pass it to the SF starter queue for each incoming message.

…he kinesis message

…into CUMULUS-359

abarciauskas-bgse · 2018-03-07T21:28:09Z

packages/api/lambdas/kinesis-consumer.js

+ * create rules.
+ *
+ * @param {*} record - input to the kinesis stream
+ * @returns {Promise} promise


Nit - could be more specific

@returns {[Promises]} Array of promises. Each promise is resolved when a message is queued for all associated kinesis rules.

yep, thanks! Think I meant to come back to that one.

abarciauskas-bgse · 2018-03-07T21:29:14Z

packages/api/lambdas/kinesis-consumer.js

 * @returns {(error|string)} Success message or error
 */
 function handler(event, context, cb) {
  const records = event.Records;

-  return Promise.all(records.map(r => processRecord(r)))
-    .then((results) => cb(null, results.filter(r => r !== undefined)))
+  return Promise.all(records.map((r) => processRecord(r)))


I think we can make this even terser - but it's a style preference:

return Promise.all(records.map(processRecord))

abarciauskas-bgse · 2018-03-07T21:30:26Z

packages/api/lambdas/kinesis-consumer.js

-  return Promise.all(records.map(r => processRecord(r)))
-    .then((results) => cb(null, results.filter(r => r !== undefined)))
+  return Promise.all(records.map((r) => processRecord(r)))
+    .then((results) => cb(null, results.filter((r) => r !== undefined)))


Another minor style thing, you can omit the parens around results when the function with one argument falls on a single line (following https://github.com/airbnb/javascript#arrows--one-arg-parens)

ESLint is complaining so I just left it.

abarciauskas-bgse · 2018-03-07T21:31:36Z

packages/api/lambdas/queue.js

@@ -1,34 +0,0 @@
-'use strict';


Nice finding some dead code and removing it! 💯

Well Alireza confirmed it was dead and Marc kinda found it, so I can only take credit for the actual physical removal.

abarciauskas-bgse · 2018-03-07T21:32:21Z

packages/ingest/queue.js

-  message.payload = { pdr };
+  message.payload = {
+    pdr
+  };


Why break this into 3 lines?

No idea.. don't remember touching that code! Fixed.

abarciauskas-bgse · 2018-03-07T21:35:10Z

packages/ingest/queue.js

+  if (message.resources) {
+    queueUrl = message.resources.queues.startSF;
+  }
+  else {


else should be on the line above, e.g.:

} else {

https://github.com/airbnb/javascript#blocks--cuddled-elses

This is removed anyway.

abarciauskas-bgse · 2018-03-07T21:35:33Z

packages/ingest/queue.js

+
+  let queueUrl = null;
+
+  if (message.resources) {


Is there documentation around when this is message.resources and when it's message.meta?

That was from when I ported it over and never removed it. So I removed that entire block.

abarciauskas-bgse · 2018-03-07T21:40:36Z

packages/api/tests/test-kinesis-consumer.js

-    ]);
-    results.Items.forEach(r => t.is(r.rule.type, 'onetime'));
-  });  
+test.serial('it should create a onetime rule for each associated workflow', async (t) => {


Do we need the serial option here? I thought we were already running tests in serial

I put that in when debugging and now have removed it. You're right, nice catch.

abarciauskas-bgse · 2018-03-07T21:42:10Z

packages/api/tests/test-kinesis-consumer.js

+    })
+  );
+
+  t.context.tableName = randomString();


Is there a reason to use random strings and regenerate these for every test? Since these are constants it makes more sense to me to define them outside of the before block and I don't really see a reason for having them be randomly generated since we statically define them in our deployments.

I was having a lot of issues with concurrency it seemed like and had the hardest time getting the queueing tests to work. This seemed to help.

I see - ya the concurrency in ava can be frustrating. I was hoping use --serial would get around these types of issues entirely. :/

abarciauskas-bgse · 2018-03-07T21:42:31Z

packages/ingest/queue.js

+ * @param {Object} event - event to queue with workflow and payload info
+ * @returns {Promise} - resolves when the message has been enqueued
+ */
+async function queueWorkflowMessage(event) {


It would be nice if we had a test for this function specifically, but wouldn't want to block this PR for that

Nope you're right and it really wasn't a big deal to add it actually. Thanks!

abarciauskas-bgse · 2018-03-07T21:44:26Z

packages/api/tests/test-kinesis-consumer.js

-    ]);
-    results.Items.forEach(r => t.is(r.rule.type, 'onetime'));
-  });  
+test.serial('it should create a onetime rule for each associated workflow', async (t) => {


Also this test description needs updating, e.g.

test('it should enqueue a message for each associated workflow', async (t) => {

Thank you!!! Fixed.

abarciauskas-bgse

Made some minor comments but none of them are really blocking, let me know when you have taken a look. Perhaps also @scisco wants to have another look.

abarciauskas-bgse · 2018-03-08T14:36:57Z

packages/api/lambdas/kinesis-consumer.js


-  return await Promise.all(oneTimeRulePromises);
+  return Rule.buildPayload(item)
+    .then((payload) => queueWorkflowMessage(payload));


Sorry missed this the first time around, this could be:

return Rule.buildPayload(item).then(queueWorkflowMessage);

Nice, thanks

abarciauskas-bgse · 2018-03-08T14:38:33Z

packages/api/lambdas/kinesis-consumer.js

- * `handler` Looks up enabled 'kinesis'-type rules associated with the collection in the event argument. It
- * creates new onetime rules for each rule found to trigger the workflow defined in the 'kinesis'-type rule.
+ * `handler` Looks up enabled 'kinesis'-type rules associated with the collection
+ * in the event argument. It creates new onetime rules for each rule found to trigger


This comment is out of date - It creates new onetime rules for each rule -> It enqueues a message for each kinesis-type rule to trigger the associated workflow.

Yep, thank you

abarciauskas-bgse · 2018-03-08T14:39:56Z

packages/api/tests/.eslintrc.json

@@ -0,0 +1,5 @@
+{
+  "rules": {
+    "no-param-reassign": "off"


Do you know off-hand why this is required? Seems like a good idea not to mutate parameters if we can avoid it.

Yeah in the tests setting t.context throws eslint errors. This eslintrc file is just for the tests.

I think we should turn it off across the repo. Reassign params in java script is a very common trend.

abarciauskas-bgse · 2018-03-08T14:42:31Z

packages/api/tests/test-kinesis-consumer.js

+
+  const model = new Rule(t.context.tableName);
+  await manager.createTable(t.context.tableName, ruleTableParams);
+  await Promise.all([rule1Params, rule2Params, disabledRuleParams]


Promise.all([rule1Params, rule2Params, disabledRuleParams].map(model.create));

I actually get all kinds of errors when I do that.

oh interesting - probably that shorthand only works for promises, not iterators. Sorry for the misleading comment.

abarciauskas-bgse · 2018-03-08T14:43:50Z

packages/ingest/test/queue.js

+  ]);
+});
+
+test('the queue receives a correctly formatted workflow message', async (t) => {


abarciauskas-bgse · 2018-03-08T14:46:14Z

packages/api/lambdas/kinesis-consumer.js

 async function processRecord(record) {
  const dataBlob = record.kinesis.data;
  const dataString = Buffer.from(dataBlob, 'base64').toString();
  const eventObject = JSON.parse(dataString);

-  await validateMessage(eventObject)
+  return validateMessage(eventObject)


If we no longer need await I don't think this function needs an async declaration

yep, thanks!

abarciauskas-bgse

Made a few more minor comments, non-blocking. Thanks @laurenfrederick !!!

scisco

👏 💯

CUMULUS-359 Add payload to one time rule created from kinesis stream

422ff3d

laurenfrederick commented Mar 5, 2018

View reviewed changes

Lauren Frederick and others added 4 commits March 5, 2018 15:44

Merge branch 'master' into CUMULUS-359

4750e3f

Merge branch 'master' into CUMULUS-359

6601642

CUMULUS-359 Update changelog

2b9f948

Merge branch 'master' into CUMULUS-359

48d3399

scisco suggested changes Mar 5, 2018

View reviewed changes

Lauren Frederick added 5 commits March 7, 2018 09:53

CUMULUS-359 Queue a workflow message instead of creating a rule for t…

9087e27

…he kinesis message

Merge branch 'CUMULUS-359' of https://github.com/cumulus-nasa/cumulus …

e1343a0

…into CUMULUS-359

CUMULUS-359 debugging

bab2cf0

CUMULUS-359 Fix tests and some cleanup

7c55d28

CUMULUS-359 Cleanup

5a5619a

abarciauskas-bgse reviewed Mar 7, 2018

View reviewed changes

Lauren Frederick added 2 commits March 8, 2018 07:30

CUMULUS-359 PR comments, test for ingest queue

186dab2

CUMULUS-359 Missed fix

67652c1

abarciauskas-bgse reviewed Mar 8, 2018

View reviewed changes

abarciauskas-bgse approved these changes Mar 8, 2018

View reviewed changes

CUMULUS-359 More PR fixes

9411a38

scisco approved these changes Mar 8, 2018

View reviewed changes

laurenfrederick merged commit a719897 into master Mar 8, 2018

scisco deleted the CUMULUS-359 branch March 8, 2018 20:32

CUMULUS-359 Add payload to one time rule created from kinesis stream #232

CUMULUS-359 Add payload to one time rule created from kinesis stream #232

Conversation

laurenfrederick commented Mar 5, 2018 • edited by marchuffnagle Loading

Changes

Test Plan

laurenfrederick left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abarciauskas-bgse Mar 7, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abarciauskas-bgse left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abarciauskas-bgse left a comment

Choose a reason for hiding this comment

scisco left a comment

Choose a reason for hiding this comment

laurenfrederick commented Mar 5, 2018 •

edited by marchuffnagle

Loading

abarciauskas-bgse Mar 7, 2018 •

edited

Loading