feat(backend): discovery function #85

NetaNir · 2021-06-15T07:36:35Z

A discovery function which triggers every 5 minutes and reads the npm couchdb _change endpoint. For every change, if the concerning package is a jsii construct, the package version tarball is copied from the npmjs registry to a staging a s3 bucket, and a message with details of the change will be posted on an SQS queue.

Every change to the registry is posted sequently to the couchdb _changes endpoint. A change example:

  {
  seq: 3299,
  id: "package_name",
  changes: [
    {
      rev: "1-81a5c9873827084aaab78b22424cfaa8",
    },
  ],
  doc: {
    _id: "package_name",
    _rev: "1-81a5c9873827084aaab78b22424cfaa8",
    name: "package_name",
    description: "...",
    "dist-tags": {
      latest: "0.0.1-beta4",
    },
    versions: {
      "0.0.1-beta1": {},
      "0.0.1-beta2": {},
      "0.0.1-beta3": {},
      "0.0.1-beta4": {}
    },
    readme: "# NOTE\nThis is beautiful README the I wrote cause I needed something for the PR, but it can have all the things markup support",
    time: {
    modified: "2021-06-15T00:59:56.624Z",
    created: "2018-02-02T03:50:11.983Z",
    "0.0.1-beta1": "2018-02-02T03:50:11.983Z",
    "0.0.1-beta2": "2018-02-02T04:04:09.226Z",
    "0.0.1-beta3": "2018-02-20T04:18:16.096Z",
    "0.0.1-beta4": "2018-02-20T06:26:18.341Z",
    },
    keywords: [
      "json",
      "schema",
    ],
    repository: {
      type: "",
      url: "",
    },
    readmeFilename: "README.md",
  },
}

see example query https://replicate.npmjs.com/registry/_changes?descending=true&limit=10&include_Docs=true

The seq is the change sequence number, the id is the package name.

A few notes on the implementation:

The _changes endpoint is designed to be consumed as a never ending stream. Since we are using a Lambda function with a fixed execution time, the function reads the latest changes in batches and exits. This is only true when reading from latest, and not in the backfill scenario, in which there are more changes than what the function can handle in its life time. In the meantime we accept that if the function times out while a batch is being processed, that batch will be processed again. This is true since that in the end of every batch processed, the latest seq is stored to the S3 bucket.
Test

Resources:

npm registry API
npm registry-follower-tutorial (from the official npm docs)

closes #3

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license

RomainMuller · 2021-06-15T07:40:47Z

src/backend/discovery/discovery.lambda.ts

+
+import * as console from 'console';


Suggested change

import * as console from 'console';

import * as console from 'console';

RomainMuller · 2021-06-15T07:41:14Z

src/backend/discovery/discovery.lambda.ts

+import * as AWS from 'aws-sdk';
+import * as Nano from 'nano';
+
+const TIMEOUT_MILLISECONDS = 10000;


Suggested change

const TIMEOUT_MILLISECONDS = 10000;

const TIMEOUT_MILLISECONDS = 10_000;

RomainMuller · 2021-06-15T07:42:00Z

src/backend/discovery/discovery.lambda.ts

+
+const TIMEOUT_MILLISECONDS = 10000;
+const CONSTRUCT_KEYWORDS = ['cdk', 'aws-cdk', 'cdk8s', 'cdktf'];
+const MARKER_FILE_NAME = 'marker';


Might want to give this a more descriptive name, e.g: npmjs-last-transaction-id or something like that.

RomainMuller · 2021-06-15T07:43:22Z

src/backend/discovery/discovery.lambda.ts

+ * npm registry API docs: https://github.com/npm/registry/blob/master/docs/REGISTRY-API.md
+ * @param context a Lambda execution context
+ */
+export async function handler(context: Context) {


Is the first argument to a lambda function not the input payload? I don't reckon the context makes it in first position...

RomainMuller · 2021-06-15T07:43:53Z

src/backend/discovery/discovery.lambda.ts

+export async function handler(context: Context) {
+  s3 = new AWS.S3();
+  sqs = new AWS.SQS();
+  functionContext = context;


I would rather not store the Context instance out of the function (it should never survive between executions).

RomainMuller · 2021-06-15T08:55:14Z

src/backend/discovery/discovery.lambda.ts

+          'Lambda-Run-Id': functionContext.awsRequestId,
+        },
+      }).promise().then(() => ok(), ko);
+    }).on('error', function (error) {


Suggested change

}).on('error', function (error) {

}).on('error', (error) => {

RomainMuller · 2021-06-15T08:55:39Z

src/backend/discovery/discovery.lambda.ts

+  }).promise().catch(function(error) {
+    throw Error(`Failed sending message to SQS\n ${error}`);
+  });


Suggested change

}).promise().catch(function(error) {

throw Error(`Failed sending message to SQS\n ${error}`);

});

}).promise();

RomainMuller · 2021-06-15T08:56:45Z

src/backend/discovery/discovery.lambda.ts

+  };
+
+  await sqs.sendMessage({
+    MessageBody: JSON.stringify(message),


Suggested change

MessageBody: JSON.stringify(message),

MessageBody: JSON.stringify(message, null, 2),

RomainMuller · 2021-06-15T08:57:12Z

src/backend/discovery/discovery.lambda.ts

+  * @param pkgJason
+  */
+function isJsiiModule(pkgJason: VersionInfo): boolean {
+  return (pkgJason?.jsii !== undefined);


Suggested change

return (pkgJason?.jsii !== undefined);

return pkgJason?.jsii != null;

RomainMuller · 2021-06-15T08:57:22Z

src/backend/discovery/discovery.lambda.ts

+  * @returns returns true if the package.json contains a jsii clause, false otherwise
+  * @param pkgJason
+  */
+function isJsiiModule(pkgJason: VersionInfo): boolean {


pkgJason 🤣

I miss him!

# Conflicts: # .projen/tasks.json # package.json # yarn.lock

Updates to the discovery function draft PR

NetaNir · 2021-06-15T21:56:41Z

src/backend/discovery/discovery.lambda.ts

 const NPM_REPLICA_REGISTRY_URL = 'https://replicate.npmjs.com/';

 /**
 * The S3 prefix for storing change that we failed processing
 */
 const FAILED_CHANGE_PREFIX = 'failed';

-let s3: AWS.S3;
-let sqs: AWS.SQS;
+let s3: AWS.S3 | undefined;


Why would we allow undefined here? The clients are required, and this only makes it so we need to add ! to every reference of s3/sqs

When we'll add unit tests on this, we'll need to be able to reset the clients so that we can use a new mock for each test. This implies allowing undefined.

One trick that @iliapolo suggested to me before is to actually vend those clients from a s3Client and sqsClient function, which lazily creates a new client if necessary... So that becomes the only place where a null-check has to exist.

NetaNir · 2021-06-15T21:57:17Z

src/backend/discovery/discovery.lambda.ts

    throw new Error('The NOTIFICATION_QUEUE_URL environment variable is not set');
  }
  queueUrl = process.env.QUEUE_URL;

-  AWS.config.update({
-    region: process.env.AWS_REGION,


This is the default behavior...

NetaNir · 2021-06-15T22:02:05Z

src/backend/discovery/discovery.lambda.ts

@@ -91,17 +89,17 @@ export async function handler(context: Context) {
        // ignores changes which are not package update, see https://github.com/cdklabs/construct-hub/issues/3#issuecomment-858246275
        .filter(change => change.doc.name)
        .map(change => {
-          return processPackageUpdate(change)
+          return processPackageUpdate(change, context)


Why not store this in a global variable same as we did for the AWS SDK clients, passing it to every function call seems cumbersome

Unlike the clients, which can be shared between executions, the context is distinct for each execution, and should not survive across executions. Saving it to a global variable is actually risky IMO.

NetaNir · 2021-06-15T22:03:06Z

src/backend/discovery/discovery.lambda.ts

+  * @returns returns true if the package.json contains a jsii clause, false otherwise
+  * @param pkgJason
+  */
+function isJsiiModule(pkgJason: VersionInfo): boolean {


I miss him!

…t-hub into neta/discovery-function

NetaNir · 2021-06-16T03:40:04Z

src/backend/discovery/discovery.lambda.ts

+  } catch (error) {
+    console.log(`Failed to load marker for bucket: ${stagingBucket}, will start read from latest`);
+  }


You are right, the cheapest thing will be to actually kill the execution. Updated.

Given that the SDK has a built in retry mechanism if we failed we should exit.

NetaNir · 2021-06-16T04:02:47Z

src/backend/discovery/discovery.lambda.ts

+        db.changesReader.stop();
+        return Promise.reject(new Error(`Error while processing batch, marker will not be updated, exiting.\n${error}`));


I have changed the .then()/.catch() style a bit, but only so it will be easier to understand the below.

Any type of error thrown from this point will cause an UnhandledPromiseRejectionWarning:

(node:78472) UnhandledPromiseRejectionWarning: Error: Error while processing batch, marker will not be updated, exiting. CredentialsError: <...>

This goes for both return Promise.reject(...) and Throw Error(). This is because both returns a Promise which is unhandled. Since this code is actually executing in the on(batch, ...)callback, the Promise needs to be resolved in the library code (I think?), which is not, causing thePromise` to go unhandled.

This does not have actual impact on the function execution as calling db.changesReader.stop() will terminate the program execution, it will just print this ugly warning to the log.

Thoughts?

@RomainMuller

…t-hub into neta/discovery-function

eladb

Missing unit tests for runtime code and construct

package.json

src/backend/discovery/index.ts

eladb · 2021-06-16T06:05:32Z

src/backend/discovery/index.ts

+  public constructor(scope: Construct, id: string, props: DiscoveryFunctionProps) {
+    super(scope, id);
+
+    const lambda = new Handler(this, 'Default', {


What's the monitoring story for this function?

eladb · 2021-06-16T06:08:23Z

src/construct-hub.ts

+    const stagingBucket = new s3.Bucket(this, 'StagingBucket', {
+      lifecycleRules: [
+        {
+          prefix: 'packages', // delete the staged tarball after 30 days


The prefix should be some const that the runtime code uses as well or passed as an environment variable

eladb · 2021-06-16T06:10:25Z

src/backend/discovery/discovery.lambda.ts

+let s3: AWS.S3 | undefined;
+let sqs: AWS.SQS | undefined;
+
+let stagingBucket: string;


Why are all of these defined outside of the function closure?

eladb · 2021-06-16T06:11:56Z

src/backend/discovery/discovery.lambda.ts

+    await s3.getObject({
+      Bucket: stagingBucket,
+      Key: MARKER_FILE_NAME,
+    }).promise().then(function(data) {
+      marker = data.Body?.toString('utf-8');
+    });


It's not a good practice to mix async with then():

Suggested change

await s3.getObject({

Bucket: stagingBucket,

Key: MARKER_FILE_NAME,

}).promise().then(function(data) {

marker = data.Body?.toString('utf-8');

});

marker = (await s3.getObject({

Bucket: stagingBucket,

Key: MARKER_FILE_NAME,

}).promise()).Body?.toString('utf-8');

});

eladb · 2021-06-16T06:12:46Z

src/backend/discovery/discovery.lambda.ts

+      marker = data.Body?.toString('utf-8');
+    });
+  } catch (error) {
+    throw new Error(`Failed to load marker for bucket: ${stagingBucket}, exiting`);


What happens the first time?

eladb · 2021-06-16T06:15:21Z

src/backend/discovery/discovery.lambda.ts

+                  'Lambda-Log-Stream': context.logStreamName,
+                  'Lambda-Run-Id': context.awsRequestId,
+                },
+              }).promise().then(() => (void undefined))));


What's going on here?

I would highly recommend avoiding resolving promises explicitly and stick to straightforward await-based style, which is easier to reason about

src/backend/discovery/discovery.lambda.ts

eladb · 2021-06-16T06:23:17Z

src/backend/discovery/discovery.lambda.ts

+async function copyPackageToS3(versionInfo: VersionInfo, key: string, context: Context): Promise<Buffer> {
+  console.log(`uploading tarball to s3, key: ${key}`);
+  return new Promise<Buffer>((ok, ko) => {
+    https.get(versionInfo.dist.tarball, (response) => {


I believe we have some nice async http library already

src/backend/discovery/discovery.lambda.ts

# Conflicts: # src/__tests__/__snapshots__/construct-hub.test.ts.snap # src/__tests__/devapp/__snapshots__/snapshot.test.ts.snap

NetaNir added 3 commits June 9, 2021 22:46

initial function implementation

4dfce75

feat: discovery function

828b582

function docs

29ee913

NetaNir marked this pull request as draft June 15, 2021 07:36

NetaNir requested a review from RomainMuller June 15, 2021 07:36

RomainMuller reviewed Jun 15, 2021

View reviewed changes

RomainMuller added 6 commits June 15, 2021 11:32

style tweaks

e9114ad

Merge remote-tracking branch 'origin/main' into neta/discovery-function

7af7269

# Conflicts: # .projen/tasks.json # package.json # yarn.lock

WIP

011108a

more tweaks

6d9bf45

add integirty gear

413651d

further out the starting index

e5f1ddc

RomainMuller force-pushed the neta/discovery-function branch 2 times, most recently from e9114ad to 29ee913 Compare June 15, 2021 16:32

Automation and others added 2 commits June 15, 2021 16:36

chore: self mutation

a771dee

Merge pull request #86 from cdklabs/rmuller/discovery-function

843cab2

Updates to the discovery function draft PR

RomainMuller approved these changes Jun 15, 2021

View reviewed changes

NetaNir commented Jun 15, 2021

View reviewed changes

NetaNir marked this pull request as ready for review June 16, 2021 03:04

NetaNir and others added 8 commits June 15, 2021 20:20

Merge branch 'main' into neta/discovery-function

1d302fe

chore: self mutation

d7ffff0

stop the stream reader on error

b322d40

Merge branch 'neta/discovery-function' of github.com:cdklabs/construc…

3559da4

…t-hub into neta/discovery-function

chore: self mutation

925c6bb

CR feedback

daa5f7f

Merge branch 'neta/discovery-function' of github.com:cdklabs/construc…

4ace344

…t-hub into neta/discovery-function

chore: self mutation

f044434

NetaNir commented Jun 16, 2021

View reviewed changes

NetaNir added 2 commits June 15, 2021 21:28

remove fallback marker

07ab543

Merge branch 'neta/discovery-function' of github.com:cdklabs/construc…

af21052

…t-hub into neta/discovery-function

chore: self mutation

f85b855

eladb reviewed Jun 16, 2021

View reviewed changes

RomainMuller force-pushed the neta/discovery-function branch from f85b855 to c72d7d7 Compare June 16, 2021 11:10

chore: self mutation

0144652

# Conflicts: # src/__tests__/__snapshots__/construct-hub.test.ts.snap # src/__tests__/devapp/__snapshots__/snapshot.test.ts.snap

RomainMuller force-pushed the neta/discovery-function branch from c72d7d7 to 0144652 Compare June 16, 2021 11:14

Automation and others added 5 commits June 16, 2021 11:18

chore: self mutation

eb50d44

tweak S3 paths

153f23e

chore: self mutation

1d45123

move default start marker

7cf4d8b

chore: self mutation

05dfd64

RomainMuller merged commit e097e11 into main Jun 16, 2021

RomainMuller deleted the neta/discovery-function branch June 16, 2021 12:15

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(backend): discovery function #85

feat(backend): discovery function #85

NetaNir commented Jun 15, 2021 •

edited

Loading

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

RomainMuller Jun 15, 2021

NetaNir Jun 15, 2021

NetaNir Jun 15, 2021

RomainMuller Jun 16, 2021

NetaNir Jun 15, 2021

RomainMuller Jun 16, 2021

NetaNir Jun 15, 2021

RomainMuller Jun 16, 2021

NetaNir Jun 15, 2021

NetaNir Jun 16, 2021

NetaNir Jun 16, 2021 •

edited

Loading

eladb left a comment

eladb Jun 16, 2021

eladb Jun 16, 2021

eladb Jun 16, 2021

eladb Jun 16, 2021

eladb Jun 16, 2021

eladb Jun 16, 2021

eladb Jun 16, 2021


	import * as console from 'console';
	import * as console from 'console';

	const TIMEOUT_MILLISECONDS = 10000;
	const TIMEOUT_MILLISECONDS = 10_000;

	}).on('error', function (error) {
	}).on('error', (error) => {

	MessageBody: JSON.stringify(message),
	MessageBody: JSON.stringify(message, null, 2),

	return (pkgJason?.jsii !== undefined);
	return pkgJason?.jsii != null;

		db.changesReader.stop();
		return Promise.reject(new Error(`Error while processing batch, marker will not be updated, exiting.\n${error}`));

feat(backend): discovery function #85

feat(backend): discovery function #85

Conversation

NetaNir commented Jun 15, 2021 • 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

NetaNir Jun 16, 2021 • edited Loading

Choose a reason for hiding this comment

eladb 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

NetaNir commented Jun 15, 2021 •

edited

Loading

NetaNir Jun 16, 2021 •

edited

Loading