feat(step-functions): add item reader path

[ChakshuGupta13]

Issue # (if applicable)

Fixes #29409

Reason for this change

• For DistributedMap state of StepFunctions, IItemReader currently only allows S3 bucket as input source to be declared statically in CDK.
• In other words, current CDK implementation only caters to static use-case where we know either bucket or bucketName (from which we can create IBucket) and pass it to IItemReader.
• Whereas via AWS Console, if we create DistributedMap manually, then we can also convey S3 source dynamically using State Input / JsonPath.
• In other words, for dynamic use-case, we will neither have bucket nor bucketName i.e. we only know state input variable which will convey bucketName e.g. $.bucketName.
• So, if we want to use IItemReader for dynamic use case also, then we will:
  • (1) need to make bucket: IBucket optional (which will be breaking - how? e.g. if some dev is currently accessing bucket via reader.bucket then dev now needs to add check for undefined)
  • (2) then add another optional fields to convey state input variable names (e.g. $.bucketName, $.key, $.prefix)
• Therefore, to avoid introducing breaking change, we can follow *Path convention of StepFunctions and introduce IItemReaderPath for dynamic use-case.
• As IItemReaderPath is being introduced to allow using dynamic S3 source for DistributedMap, hence, we will also need to convey about it and its usage example via README.
• Following proposal discussed briefly with @GavinZZ:

Description of changes

• Add IItemReaderPath interface (and its pros interface)
• Add S3ObjectsItemReaderPath as one of many concrete classes of IItemReaderPath - this class also helps with unit-testing and integration-testing.
• Modify index to export new constructs
• Modify DistributedMap (and its props) to allow itemReaderPath? (which will be mutually exclusive with itemReader? and itemsPath?) and utilise it for render
• Add when and how IItemReaderPath shall be used in README.

Description of how you validated changes

• Via new unit-tests for DistributedMap (via yarn build+test)
• Via new integration test for S3ObjectsItemReaderPath (with snapshot created)
  • Via yarn build --directory test/aws-stepfunctions/test && yarn integ test/aws-stepfunctions/test/integ.item-reader-path-s3-object.js && yarn integ-runner --update-on-failed
  • Verified expected step function execution result during snapshot creation
• Previewed README changes.

Checklist

• My code adheres to the CONTRIBUTING GUIDE and DESIGN GUIDELINES

---

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

[aws-cdk-automation]

The pull request linter has failed. See the aws-cdk-automation comment below for failure reasons. If you believe this pull request should receive an exemption, please comment and provide a justification.

A comment requesting an exemption should contain the text Exemption Request. Additionally, if clarification is needed add Clarification Request to a comment.

✅ Updated pull request passes all PRLinter validations. Dismissing previous PRLinter review.

[GavinZZ]

Generally looks good. Just some questions to help me understand.

[GavinZZ]

Is this comment needed and correct? Shouldn't it bucketNamePath?

[ChakshuGupta13]

• About necessity of comment, yes, I thought this construct's usage is not very straightforward, hence, developers may need to do try-and-error with state input if this comment is not present, hence, to save time of developers by avoiding try-and-error, we can convey how state input shall look like.
• bucketNamePath is a parameter of S3ObjectsItemReaderPath which contains key of state input field (here bucketName) which contains actual value ('my-bucket'), so, no, it shouldn't be bucketNamePath unless (bucketNamePath: '$. bucketNamePath' in S3ObjectsItemReaderPath), hence, yes, comment is correct. (Note: Same is verified via integration test implementation)

new sfn.S3ObjectsItemReaderPath({
    bucketNamePath: '$.bucketName', /** ~ bucketNamePath: '$.X' */
    ...
  }),

/**
 * State Input:
 * {
 *   bucketName: 'my-bucket', // ~ X: 'my-bucket'
 *   ...
 * }
 */

[GavinZZ]

That's good to know. I also deployed this template and you're right. I think this is an important piece of information. I would like to request you to format the README file a bit clearer. Can we split this part into subheaders, one for static bucket and one for dynamic. In the dynamic add an (complete) example with the stack definition and state input.

[ChakshuGupta13]

Is following content for README okay?

• Sub-header at level 4 heading were too small to make a difference and also text content of each heading shall be some construct like DistributedMap, hence, didn't utilise further sub-headers instead explained usage sequentially.
• Ask: If this format seem alright, then will revise PR and in upcoming PRs will add examples for remaining input source types as they are not currently present for dynamic use-case.

---

DistributedMap supports various input source types to determine an array to iterate over:

• JSON array from the state input
  • By default, DistributedMap assumes whole state input is an JSON array and iterates over it:

  [
    "item1",
    "item2"
  ]

  const distributedMap = new sfn.DistributedMap(this, 'DistributedMap');
  distributedMap.itemProcessor(new sfn.Pass(this, 'Pass'));

  • When input source is present at a specific path in state input, then itemsPath can be utilised to configure the iterator source.

  {
    "distributedMapItemList": [
      "item1",
      "item2"
    ]
  }

  const distributedMap = new sfn.DistributedMap(this, 'DistributedMap', {
    itemsPath: '$.distributedMapItemList',
  });
  distributedMap.itemProcessor(new sfn.Pass(this, 'Pass'));

• Objects in a S3 bucket and optional prefix
  • When DistributedMap is required to iterate over objects stored in a S3 bucket, then if required parameters like bucket are known while creating StateMachine (statically or at compile time), then an object of S3ObjectsItemReader (implementing IItemReader) can be passed to itemReader to configure the iterator source.

  my-bucket
  |
  +--item1
  |
  +--otherItem
  |
  +--item2
  |
  ...

  import * as s3 from 'aws-cdk-lib/aws-s3';
  
  const bucket = new s3.Bucket(this, 'Bucket', {
    bucketName: 'my-bucket',
  });
  
  const distributedMap = new sfn.DistributedMap(this, 'DistributedMap', {
    itemReader: new sfn.S3ObjectsItemReader({
      bucket,
      prefix: 'item',
    }),
  });
  distributedMap.itemProcessor(new sfn.Pass(this, 'Pass'));

  • But if required parameters like bucketName are only known while starting execution of StateMachine (dynamically or at run-time) via state input, then an object of S3ObjectsItemReaderPath (implementing IItemReaderPath) can be passed to itemReaderPath to configure the iterator source.

  {
    "bucketName": "my-bucket",
    "prefix": "item"
  }

  const distributedMap = new sfn.DistributedMap(this, 'DistributedMap', {
    itemReaderPath: new sfn.S3ObjectsItemReaderPath({
      bucketNamePath: '$.bucketName',
      prefixPath: '$.prefix',
    }),
  });
  distributedMap.itemProcessor(new sfn.Pass(this, 'Pass'));

  • Both itemReader and itemReaderPath are mutually exclusive. For example, if bucket is known at compile time but prefix is only known at run-time, then both cannot be used simultaneously.
• JSON array in a JSON file stored in S3
  • When DistributedMap is required to iterate over a JSON array stored in a JSON file in a S3 bucket, then if required parameters like bucket are known while creating StateMachine (statically or at compile time), then an object of S3JsonItemReader (implementing IItemReader) can be passed to itemReader to configure the iterator source.

  my-bucket
  |
  +--input.json
  |
  ...

  [
    "item1",
    "item2"
  ]

  import * as s3 from 'aws-cdk-lib/aws-s3';
  
  const bucket = new s3.Bucket(this, 'Bucket', {
    bucketName: 'my-bucket',
  });
  
  const distributedMap = new sfn.DistributedMap(this, 'DistributedMap', {
    itemReader: new sfn.S3JsonItemReader({
      bucket,
      key: 'input.json',
    }),
  });
  distributedMap.itemProcessor(new sfn.Pass(this, 'Pass'));

• CSV file stored in S3
• S3 inventory manifest stored in S3

Map states in Distributed mode also support writing results of the iterator to an S3 bucket and optional prefix. Use a ResultWriter object provided via the optional resultWriter property to configure which S3 location iterator results will be written. The default behavior id resultWriter is omitted is to use the state output payload. However, if the iterator results are larger than the 256 kb limit for Step Functions payloads then the State Machine will fail.

import * as s3 from 'aws-cdk-lib/aws-s3';

// create a bucket
const bucket = new s3.Bucket(this, 'Bucket');

const distributedMap = new sfn.DistributedMap(this, 'Distributed Map State', {
  resultWriter: new sfn.ResultWriter({
    bucket: bucket,
    prefix: 'my-prefix',
  })
});
distributedMap.itemProcessor(new sfn.Pass(this, 'Pass State'));

...

[GavinZZ]

I think the code changes are good to go. I would like to request you to update the README clearer with subheaders and full examples. I'm happy to approve once it's updated.

[moelasmar]

I am finding this approach very confusing to customers. Also we are duplicating every thing in 2 versions of almost the same classes, which means that in future for any updates to these classes, we are duplicating the work. I see that this may be some source of errors.

The main difference I see between the ItemReader and ItemReaderPath is for the first one, we automatically add the policy to access the S3 bucket, and the other one is we could not add this policy. We need to think about a better design approach.

[moelasmar]

one of the options is to deprecate the current ItemReader class, and add a new one, but I also do not like the idea of keep deprecating existing properties for new properties and ask customers to update their code.

[ChakshuGupta13]

Following is intentional, with IItemReaderPath, during compile time, we will not have information about bucket, hence, we can't add any permission during compile time, therefore, this action needs to be taken separately by caller who starts StateMachine execution with bucketName. Even, if we consider adding * policy statement by default here, then it will violate minimum access to resource guideline.

The main difference I see between the ItemReader and ItemReaderPath is for the first one, we automatically add the policy to access the S3 bucket, and the other one is we could not add this policy.

Following is precisely issue with current codebase that it is not extensible to allow changes and most of such options are already considered to decent extent (refer PR description) - and among those, this approach was least troublesome.

We need to think about a better design approach.
one of the options is to deprecate the current ItemReader class, and add a new one, but I also do not like the idea of keep deprecating existing properties for new properties and ask customers to update their code.

If you have any other alternatives, please feel free to share - happy to ponder over and utilise them. This is exactly why this PR is raised with minimum number of constructs i.e. to gain consensus.

[ChakshuGupta13]

Similar duplicate parameters are also utilised in different places (reference) - although they have simple types, not full-fledged classes (or interfaces)

  /**
   * Error code used to represent this failure
   *
   * @default - No error code
   */
  readonly error?: string;

  /**
   * JsonPath expression to select part of the state to be the error to this state.
   *
   * You can also use an intrinsic function that returns a string to specify this property.
   * The allowed functions include States.Format, States.JsonToString, States.ArrayGetItem, States.Base64Encode, States.Base64Decode, States.Hash, and States.UUID.
   *
   * @default - No error path
   */
  readonly errorPath?: string;

I am finding this approach very confusing to customers. Also we are duplicating every thing in 2 versions of almost the same classes

[ChakshuGupta13]

Original proposal required breaking changes but avoided code duplication, hence also asked in ticket:

[GavinZZ]

@moelasmar thanks for a second review and provide the feedback. I do agree that a number of resources are duplicate between IItemReader and IIteamReaderPath. This is not the best design but as mentioned earlier, we can't directly modify IItemReader due to breaking changes. Specifically IItemReader takes a required parameter bucket: IBucket and this doesn't work for dynamic parameter handling.

I agree this is not the best practice but there are many places in CDK that has similar issues. Would like you get your thought on the next step. One thing I can think of is to create a base interface that defines the shared parameters and IItemReader and IItemReaderPath to extend the interface and has its own definitions from there.

[ChakshuGupta13]

I tried to think about base interface as we discussed briefly (re:Slack & WorkLog) but then got stuck as to how will that be better useful? We will still need different implementations (both for static and dynamic) for each reader type (object, JSON, CSV, manifest etc.)

One thing I can think of is to create a base interface that defines the shared parameters and IItemReader and IItemReaderPath to extend the interface and has its own definitions from there.

So, can you please share if you foresee any benefit with base class (and if possible, can you please explain "how" like an example - that will help clarify few thoughts for me).

---

Secondly, I was thinking about examples like following:

 /**
   * Error code used to represent this failure
   *
   * @default - No error code
   */
  readonly error?: string;

  /**
   * JsonPath expression to select part of the state to be the error to this state.
   *
   * You can also use an intrinsic function that returns a string to specify this property.
   * The allowed functions include States.Format, States.JsonToString, States.ArrayGetItem, States.Base64Encode, States.Base64Decode, States.Hash, and States.UUID.
   *
   * @default - No error path
   */
  readonly errorPath?: string;

In such example, we always have two parameters (one for static and one for dynamic), so, now if IItemReader and its implementations have X number of static parameters (e.g. bucket, prefix, key etc.), then to have dynamic alternatives, we will have same number of parameters (e.g. bucketNamePath, prefixPath, keyPath etc.).

Also:

• render method for both use-cases will differ (e.g. for dynamic Parameters will be suffixed with .$)
• As mentioned earlier, for dynamic use-case, we cannot have policyStatements defined statically since required information is unavailable at that time, hence, that will also differ.

Therefore, only thing common between static and dynamic alternatives that can be de-duplicated is resource which is initialised to an Arn in constructor, hence, by making base class (or trying to generalise these classes), we can only avoid duplication of resource at max (since, we need do parameters like bucketNamePath, prefixPath, keyPath etc.)

---

Ideal solution would be to have classes like following:

export interface IItemReader {
  /** bucketName and bucketNamePath mutually exclusive*/
  readonly bucketName?: string;
  readonly bucketNamePath?: string;

  readonly resource: string;
  readonly maxItems?: number;
  render(): any;
}

export class S3ObjectsItemReader implements IItemReader {
  readonly bucketName?: string;
  readonly bucketNamePath?: string;

  /** prefix and prefixPath mutually exclusive*/
  readonly prefix?: string;
  readonly prefixPath?: string;

  readonly resource: string;
  readonly maxItems?: number;

  constructor(props...) {
    this.bucketName = props.bucketName;
    this.bucketNamePath = props.bucketNamePath;
    this.prefix = props.prefix;
    this.prefixPath = props.prefixPath;
    this.maxItems = props.maxItems;
    this.resource = Arn.format({
      region: '',
      account: '',
      partition: Aws.PARTITION,
      service: 'states',
      resource: 's3',
      resourceName: 'listObjectsV2',
      arnFormat: ArnFormat.COLON_RESOURCE_NAME,
    });
  }

  public render(): any {
    return FieldUtils.renderObject({
      Resource: this.resource,
      ...(this.maxItems && { ReaderConfig: { MaxItems: this.maxItems } }),
      Parameters: {
        ... (this.bucketName && { Bucket: this.bucketName } ),
        ... (this.bucketNamePath && { 'Bucket.$': this.bucketNamePath } ),
        ...(this.prefix && { Prefix: this.prefix }),
        ...(this.prefixPath && { 'Prefix.$': this.prefixPath }),
      },
    });
  }
}

This way we can even have partial static parameters and rest dynamic - example:

const distributedMap = new sfn.DistributedMap(this, 'DistributedMap', {
    itemReaderPath: new sfn.S3ObjectsItemReader({
      bucketName: 'my-bucket',
      prefixPath: '$.prefix',
    }),
  });
  distributedMap.itemProcessor(new sfn.Pass(this, 'Pass'));

As much as I like this freedom to choose which parameter needs to be dynamic, this will need breaking changes.

EDIT: Forgot that validations (i.e. mutually exclusiveness above) can only be done for States - not state-less (node-less) constructs like implementations of IItemReader, so, above solution may not be feasible also (unless these are made actual Construct which poses different problems altogether like assigning unique logical IDs etc.) - leaving us with two different classes only.

[ChakshuGupta13]

@moelasmar @GavinZZ
my colleague discovered following way with which we can provide dynamic parameters (except for Bucket because it is currently of type IBucket instead of string):

const myBucket = new Bucket(stack, 'MyBucket', ...);
const distributedMap = new sfn.DistributedMap(this, 'DistributedMap', {
    itemReader: new sfn.S3ObjectsItemReader({
      bucket: myBucket,
      prefix: JsonPath.stringAt('$.prefix'),
    }),
  });
  distributedMap.itemProcessor(new sfn.Pass(this, 'Pass'));

This code snippet produces following cdk.out template:

...
"Parameters\":{\"Bucket\":\"",
       {
        "Ref": "MyBucket..."
       },
       "\",\"Prefix.$\":\"$.prefix\"}}}}}"
      ]
...

Not sure why this works but since it works, this PR may not be needed for the said half issue (to be able to pass dynamic file name) but for other half issue i.e. how to also allow Bucket to be dynamic? For that, the ideal would be breaking change where bucket: IBucket is changed to bucket: string.

[GavinZZ]

Nice, but we still need to fix the issue where it only takes IBucket. I don't find much values actually to move the common parameters into a base interface and have IItemReader and IItemReaderPath to extend it.

@moelasmar would you mind take another look and provide some feedback if you have any ideas on a non-breaking change but don't have to duplicate the same properties.

[moelasmar]

why you did not do similar thing for S3ManifestItemReader, S3CsvItemReader, and S3JsonItemReader

[ChakshuGupta13]

They will be taken in follow up PRs for following reasons:

• Confirmation is first needed whether these type of changes are acceptable.
• It is not good to raise one huge PR with all changes when that huge PR can be easily broken down into smaller chunks.

[moelasmar]

should we also validate that there is no policy automatically added for this bucket.

[ChakshuGupta13]

Why? There is no policyStatement parameter with IItemReaderPath unlike IItemReader.

By this logic, we will start checking for absence of everything everywhere just because we know that something similar existed somewhere else.

Second, no real bucket exists for this test - we are just passing state input path which will convey bucketName dynamically, hence, checking whether permissions are added will not be possible anyway.

[aws-cdk-automation]

AWS CodeBuild CI Report

• CodeBuild project: AutoBuildv2Project1C6BFA3F-wQm2hXv2jqQv
• Commit ID: a4043d2
• Result: SUCCEEDED
• Build Logs (available for 30 days)

Powered by github-codebuild-logs, available on the AWS Serverless Application Repository

[GavinZZ]

Pinging @moelasmar for a second review for any blocking feedbacks.