Namespace Amazon.CDK.AWS.StepFunctions

AWS Step Functions Construct Library

The aws-cdk-lib/aws-stepfunctions package contains constructs for building serverless workflows using objects. Use this in conjunction with the aws-cdk-lib/aws-stepfunctions-tasks package, which contains classes used to call other AWS services.

Defining a workflow looks like this (for the Step Functions Job Poller example):

Example

using Amazon.CDK.AWS.Lambda;

Function submitLambda;
Function getStatusLambda;


var submitJob = new LambdaInvoke(this, "Submit Job", new LambdaInvokeProps {
    LambdaFunction = submitLambda,
    // Lambda's result is in the attribute `guid`
    OutputPath = "$.guid"
});

var waitX = new Wait(this, "Wait X Seconds", new WaitProps {
    Time = WaitTime.SecondsPath("$.waitSeconds")
});

var getStatus = new LambdaInvoke(this, "Get Job Status", new LambdaInvokeProps {
    LambdaFunction = getStatusLambda,
    // Pass just the field named "guid" into the Lambda, put the
    // Lambda's result in a field called "status" in the response
    InputPath = "$.guid",
    OutputPath = "$.status"
});

var jobFailed = new Fail(this, "Job Failed", new FailProps {
    Cause = "AWS Batch Job Failed",
    Error = "DescribeJob returned FAILED"
});

var finalStatus = new LambdaInvoke(this, "Get Final Job Status", new LambdaInvokeProps {
    LambdaFunction = getStatusLambda,
    // Use "guid" field as input
    InputPath = "$.guid",
    OutputPath = "$.Payload"
});

var definition = submitJob.Next(waitX).Next(getStatus).Next(new Choice(this, "Job Complete?").When(Condition.StringEquals("$.status", "FAILED"), jobFailed).When(Condition.StringEquals("$.status", "SUCCEEDED"), finalStatus).Otherwise(waitX));

new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition),
    Timeout = Duration.Minutes(5),
    Comment = "a super cool state machine"
});

You can find more sample snippets and learn more about the service integrations in the aws-cdk-lib/aws-stepfunctions-tasks package.

State Machine

A stepfunctions.StateMachine is a resource that takes a state machine definition. The definition is specified by its start state, and encompasses all states reachable from the start state:

var startState = new Pass(this, "StartState");

new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(startState)
});

State machines are made up of a sequence of Steps, which represent different actions taken in sequence. Some of these steps represent control flow (like Choice, Map and Wait) while others represent calls made against other AWS services (like LambdaInvoke). The second category are called Tasks and they can all be found in the module aws-stepfunctions-tasks.

State machines execute using an IAM Role, which will automatically have all permissions added that are required to make all state machine tasks execute properly (for example, permissions to invoke any Lambda functions you add to your workflow). A role will be created by default, but you can supply an existing one as well.

Set the removalPolicy prop to RemovalPolicy.RETAIN if you want to retain the execution history when CloudFormation deletes your state machine.

Alternatively you can specify an existing step functions definition by providing a string or a file that contains the ASL JSON.

new StateMachine(this, "StateMachineFromString", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromString("{\"StartAt\":\"Pass\",\"States\":{\"Pass\":{\"Type\":\"Pass\",\"End\":true}}}")
});

new StateMachine(this, "StateMachineFromFile", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromFile("./asl.json")
});

State Machine Data

An Execution represents each time the State Machine is run. Every Execution has State Machine Data: a JSON document containing keys and values that is fed into the state machine, gets modified by individual steps as the state machine progresses, and finally is produced as output.

By default, the entire Data object is passed into every state, and the return data of the step becomes new the new Data object. This behavior can be modified by supplying values for inputPath, resultSelector, resultPath and outputPath.

Manipulating state machine data using inputPath, resultSelector, resultPath and outputPath

These properties impact how each individual step interacts with the state machine data:

Their values should be a string indicating a JSON path into the State Machine Data object (like "$.MyKey"). If absent, the values are treated as if they were "$", which means the entire object.

The following pseudocode shows how AWS Step Functions uses these parameters when executing a step:

// Schematically show how Step Functions evaluates functions.
// [] represents indexing into an object by a using JSON path.

input = state[inputPath]

result = invoke_step(select_parameters(input))

state[resultPath] = result[resultSelector]

state = state[outputPath]

Instead of a JSON path string, each of these paths can also have the special value JsonPath.DISCARD, which causes the corresponding indexing expression to return an empty object ({}). Effectively, that means there will be an empty input object, an empty result object, no effect on the state, or an empty state, respectively.

Some steps (mostly Tasks) have Parameters, which are selected differently. See the next section.

See the official documentation on input and output processing in Step Functions.

Passing Parameters to Tasks

Tasks take parameters, whose values can be taken from the State Machine Data object. For example, your workflow may want to start a CodeBuild with an environment variable that is taken from the State Machine data, or pass part of the State Machine Data into an AWS Lambda Function.

In the original JSON-based states language used by AWS Step Functions, you would add .$ to the end of a key to indicate that a value needs to be interpreted as a JSON path. In the CDK API you do not change the names of any keys. Instead, you pass special values. There are 3 types of task inputs to consider:

For example, to pass the value that's in the data key of OrderId to a Lambda function as you invoke it, use JsonPath.stringAt('$.OrderId'), like so:

using Amazon.CDK.AWS.Lambda;

Function orderFn;


var submitJob = new LambdaInvoke(this, "InvokeOrderProcessor", new LambdaInvokeProps {
    LambdaFunction = orderFn,
    Payload = TaskInput.FromObject(new Dictionary<string, object> {
        { "OrderId", JsonPath.StringAt("$.OrderId") }
    })
});

The following methods are available:

Method	Purpose
JsonPath.stringAt('$.Field')	reference a field, return the type as a string.
JsonPath.listAt('$.Field')	reference a field, return the type as a list of strings.
JsonPath.numberAt('$.Field')	reference a field, return the type as a number. Use this for functions that expect a number argument.
JsonPath.objectAt('$.Field')	reference a field, return the type as an IResolvable. Use this for functions that expect an object argument.
JsonPath.entirePayload	reference the entire data object (equivalent to a path of $).
JsonPath.taskToken	reference the Task Token, used for integration patterns that need to run for a long time.
JsonPath.executionId	reference the Execution Id field of the context object.
JsonPath.executionInput	reference the Execution Input object of the context object.
JsonPath.executionName	reference the Execution Name field of the context object.
JsonPath.executionRoleArn	reference the Execution RoleArn field of the context object.
JsonPath.executionStartTime	reference the Execution StartTime field of the context object.
JsonPath.stateEnteredTime	reference the State EnteredTime field of the context object.
JsonPath.stateName	reference the State Name field of the context object.
JsonPath.stateRetryCount	reference the State RetryCount field of the context object.
JsonPath.stateMachineId	reference the StateMachine Id field of the context object.
JsonPath.stateMachineName	reference the StateMachine Name field of the context object.

You can also call intrinsic functions using the methods on JsonPath:

Method	Purpose
JsonPath.array(JsonPath.stringAt('$.Field'), ...)	make an array from other elements.
JsonPath.arrayPartition(JsonPath.listAt('$.inputArray'), 4)	partition an array.
JsonPath.arrayContains(JsonPath.listAt('$.inputArray'), 5)	determine if a specific value is present in an array.
JsonPath.arrayRange(1, 9, 2)	create a new array containing a specific range of elements.
JsonPath.arrayGetItem(JsonPath.listAt('$.inputArray'), 5)	get a specified index's value in an array.
JsonPath.arrayLength(JsonPath.listAt('$.inputArray'))	get the length of an array.
JsonPath.arrayUnique(JsonPath.listAt('$.inputArray'))	remove duplicate values from an array.
JsonPath.base64Encode(JsonPath.stringAt('$.input'))	encode data based on MIME Base64 encoding scheme.
JsonPath.base64Decode(JsonPath.stringAt('$.base64'))	decode data based on MIME Base64 decoding scheme.
JsonPath.hash(JsonPath.objectAt('$.Data'), JsonPath.stringAt('$.Algorithm'))	calculate the hash value of a given input.
JsonPath.jsonMerge(JsonPath.objectAt('$.Obj1'), JsonPath.objectAt('$.Obj2'))	merge two JSON objects into a single object.
JsonPath.stringToJson(JsonPath.stringAt('$.ObjStr'))	parse a JSON string to an object
JsonPath.jsonToString(JsonPath.objectAt('$.Obj'))	stringify an object to a JSON string
JsonPath.mathRandom(1, 999)	return a random number.
JsonPath.mathAdd(JsonPath.numberAt('$.value1'), JsonPath.numberAt('$.step'))	return the sum of two numbers.
JsonPath.stringSplit(JsonPath.stringAt('$.inputString'), JsonPath.stringAt('$.splitter'))	split a string into an array of values.
JsonPath.uuid()	return a version 4 universally unique identifier (v4 UUID).
JsonPath.format('The value is {}.', JsonPath.stringAt('$.Value'))	insert elements into a format string.

Amazon States Language

This library comes with a set of classes that model the Amazon States Language. The following State classes are supported:

An arbitrary JSON object (specified at execution start) is passed from state to state and transformed during the execution of the workflow. For more information, see the States Language spec.

Task

A Task represents some work that needs to be done. Do not use the Task class directly.

Instead, use one of the classes in the aws-cdk-lib/aws-stepfunctions-tasks module, which provide a much more ergonomic way to integrate with various AWS services.

Pass

A Pass state passes its input to its output, without performing work. Pass states are useful when constructing and debugging state machines.

The following example injects some fixed data into the state machine through the result field. The result field will be added to the input and the result will be passed as the state's output.

// Makes the current JSON state { ..., "subObject": { "hello": "world" } }
var pass = new Pass(this, "Add Hello World", new PassProps {
    Result = Result.FromObject(new Dictionary<string, object> { { "hello", "world" } }),
    ResultPath = "$.subObject"
});

// Set the next state
var nextState = new Pass(this, "NextState");
pass.Next(nextState);

The Pass state also supports passing key-value pairs as input. Values can be static, or selected from the input with a path.

The following example filters the greeting field from the state input and also injects a field called otherData.

var pass = new Pass(this, "Filter input and inject data", new PassProps {
    StateName = "my-pass-state",  // the custom state name for the Pass state, defaults to 'Filter input and inject data' as the state name
    Parameters = new Dictionary<string, object> {  // input to the pass state
        { "input", JsonPath.StringAt("$.input.greeting") },
        { "otherData", "some-extra-stuff" } }
});

The object specified in parameters will be the input of the Pass state. Since neither Result nor ResultPath are supplied, the Pass state copies its input through to its output.

Learn more about the Pass state

Wait

A Wait state waits for a given number of seconds, or until the current time hits a particular time. The time to wait may be taken from the execution's JSON state.

// Wait until it's the time mentioned in the the state object's "triggerTime"
// field.
var wait = new Wait(this, "Wait For Trigger Time", new WaitProps {
    Time = WaitTime.TimestampPath("$.triggerTime")
});

// Set the next state
var startTheWork = new Pass(this, "StartTheWork");
wait.Next(startTheWork);

Choice

A Choice state can take a different path through the workflow based on the values in the execution's JSON state:

var choice = new Choice(this, "Did it work?");

// Add conditions with .when()
var successState = new Pass(this, "SuccessState");
var failureState = new Pass(this, "FailureState");
choice.When(Condition.StringEquals("$.status", "SUCCESS"), successState);
choice.When(Condition.NumberGreaterThan("$.attempts", 5), failureState);

// Use .otherwise() to indicate what should be done if none of the conditions match
var tryAgainState = new Pass(this, "TryAgainState");
choice.Otherwise(tryAgainState);

If you want to temporarily branch your workflow based on a condition, but have all branches come together and continuing as one (similar to how an if ... then ... else works in a programming language), use the .afterwards() method:

var choice = new Choice(this, "What color is it?");
var handleBlueItem = new Pass(this, "HandleBlueItem");
var handleRedItem = new Pass(this, "HandleRedItem");
var handleOtherItemColor = new Pass(this, "HanldeOtherItemColor");
choice.When(Condition.StringEquals("$.color", "BLUE"), handleBlueItem);
choice.When(Condition.StringEquals("$.color", "RED"), handleRedItem);
choice.Otherwise(handleOtherItemColor);

// Use .afterwards() to join all possible paths back together and continue
var shipTheItem = new Pass(this, "ShipTheItem");
choice.Afterwards().Next(shipTheItem);

You can add comments to Choice states as well as conditions that use choice.when.

var choice = new Choice(this, "What color is it?", new ChoiceProps {
    Comment = "color comment"
});
var handleBlueItem = new Pass(this, "HandleBlueItem");
var handleOtherItemColor = new Pass(this, "HanldeOtherItemColor");
choice.When(Condition.StringEquals("$.color", "BLUE"), handleBlueItem, new ChoiceTransitionOptions {
    Comment = "blue item comment"
});
choice.Otherwise(handleOtherItemColor);

If your Choice doesn't have an otherwise() and none of the conditions match the JSON state, a NoChoiceMatched error will be thrown. Wrap the state machine in a Parallel state if you want to catch and recover from this.

Available Conditions

see step function comparison operators

Parallel

A Parallel state executes one or more subworkflows in parallel. It can also be used to catch and recover from errors in subworkflows.

var parallel = new Parallel(this, "Do the work in parallel");

// Add branches to be executed in parallel
var shipItem = new Pass(this, "ShipItem");
var sendInvoice = new Pass(this, "SendInvoice");
var restock = new Pass(this, "Restock");
parallel.Branch(shipItem);
parallel.Branch(sendInvoice);
parallel.Branch(restock);

// Retry the whole workflow if something goes wrong with exponential backoff
parallel.AddRetry(new RetryProps {
    MaxAttempts = 1,
    MaxDelay = Duration.Seconds(5),
    JitterStrategy = JitterType.FULL
});

// How to recover from errors
var sendFailureNotification = new Pass(this, "SendFailureNotification");
parallel.AddCatch(sendFailureNotification);

// What to do in case everything succeeded
var closeOrder = new Pass(this, "CloseOrder");
parallel.Next(closeOrder);

Succeed

Reaching a Succeed state terminates the state machine execution with a successful status.

var success = new Succeed(this, "We did it!");

Fail

Reaching a Fail state terminates the state machine execution with a failure status. The fail state should report the reason for the failure. Failures can be caught by encompassing Parallel states.

var fail = new Fail(this, "Fail", new FailProps {
    Error = "WorkflowFailure",
    Cause = "Something went wrong"
});

The Fail state also supports returning dynamic values as the error and cause that are selected from the input with a path.

var fail = new Fail(this, "Fail", new FailProps {
    ErrorPath = JsonPath.StringAt("$.someError"),
    CausePath = JsonPath.StringAt("$.someCause")
});

You can also use an intrinsic function that returns a string to specify CausePath and ErrorPath. The available functions include States.Format, States.JsonToString, States.ArrayGetItem, States.Base64Encode, States.Base64Decode, States.Hash, and States.UUID.

var fail = new Fail(this, "Fail", new FailProps {
    ErrorPath = JsonPath.Format("error: {}.", JsonPath.StringAt("$.someError")),
    CausePath = "States.Format('cause: {}.', $.someCause)"
});

Map

A Map state can be used to run a set of steps for each element of an input array. A Map state will execute the same steps for multiple entries of an array in the state input.

While the Parallel state executes multiple branches of steps using the same input, a Map state will execute the same steps for multiple entries of an array in the state input.

var map = new Map(this, "Map State", new MapProps {
    MaxConcurrency = 1,
    ItemsPath = JsonPath.StringAt("$.inputForMap"),
    ItemSelector = new Dictionary<string, object> {
        { "item", JsonPath.StringAt("$.Map.Item.Value") }
    },
    ResultPath = "$.mapOutput"
});

// The Map iterator can contain a IChainable, which can be an individual or multiple steps chained together.
// Below example is with a Choice and Pass step
var choice = new Choice(this, "Choice");
var condition1 = Condition.StringEquals("$.item.status", "SUCCESS");
var step1 = new Pass(this, "Step1");
var step2 = new Pass(this, "Step2");
var finish = new Pass(this, "Finish");

var definition = choice.When(condition1, step1).Otherwise(step2).Afterwards().Next(finish);

map.ItemProcessor(definition);

To define a distributed Map state set itemProcessors mode to ProcessorMode.DISTRIBUTED. An executionType must be specified for the distributed Map workflow.

var map = new Map(this, "Map State", new MapProps {
    MaxConcurrency = 1,
    ItemsPath = JsonPath.StringAt("$.inputForMap"),
    ItemSelector = new Dictionary<string, object> {
        { "item", JsonPath.StringAt("$.Map.Item.Value") }
    },
    ResultPath = "$.mapOutput"
});

map.ItemProcessor(new Pass(this, "Pass State"), new ProcessorConfig {
    Mode = ProcessorMode.DISTRIBUTED,
    ExecutionType = ProcessorType.STANDARD
});

Visit <a href="https://docs.aws.amazon.com/step-functions/latest/dg/use-dist-map-orchestrate-large-scale-parallel-workloads.html">Using Map state in Distributed mode to orchestrate large-scale parallel workloads</a> for more details.

Distributed Map

Step Functions provides a high-concurrency mode for the Map state known as Distributed mode. In this mode, the Map state can accept input from large-scale Amazon S3 data sources. For example, your input can be a JSON or CSV file stored in an Amazon S3 bucket, or a JSON array passed from a previous step in the workflow. A Map state set to Distributed is known as a Distributed Map state. In this mode, the Map state runs each iteration as a child workflow execution, which enables high concurrency of up to 10,000 parallel child workflow executions. Each child workflow execution has its own, separate execution history from that of the parent workflow.

Use the Map state in Distributed mode when you need to orchestrate large-scale parallel workloads that meet any combination of the following conditions:

A DistributedMap state can be used to run a set of steps for each element of an input array with high concurrency. A DistributedMap state will execute the same steps for multiple entries of an array in the state input or from S3 objects.

var distributedMap = new DistributedMap(this, "Distributed Map State", new DistributedMapProps {
    MaxConcurrency = 1,
    ItemsPath = JsonPath.StringAt("$.inputForMap")
});
distributedMap.ItemProcessor(new Pass(this, "Pass State"));

DistributedMap supports various input source types to determine a list of objects to iterate over:

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.

using Amazon.CDK.AWS.S3;


// create a bucket
var bucket = new Bucket(this, "Bucket");

var distributedMap = new DistributedMap(this, "Distributed Map State", new DistributedMapProps {
    ResultWriter = new ResultWriter(new ResultWriterProps {
        Bucket = bucket,
        Prefix = "my-prefix"
    })
});
distributedMap.ItemProcessor(new Pass(this, "Pass State"));

If you want to specify the execution type for the ItemProcessor in the DistributedMap, you must set the mapExecutionType property in the DistributedMap class. When using the DistributedMap class, the ProcessorConfig.executionType property is ignored.

In the following example, the execution type for the ItemProcessor in the DistributedMap is set to EXPRESS based on the value specified for mapExecutionType.

var distributedMap = new DistributedMap(this, "DistributedMap", new DistributedMapProps {
    MapExecutionType = StateMachineType.EXPRESS
});

distributedMap.ItemProcessor(new Pass(this, "Pass"), new ProcessorConfig {
    Mode = ProcessorMode.DISTRIBUTED,
    ExecutionType = ProcessorType.STANDARD
});

Custom State

It's possible that the high-level constructs for the states or stepfunctions-tasks do not have the states or service integrations you are looking for. The primary reasons for this lack of functionality are:

If a feature is not available, a CustomState can be used to supply any Amazon States Language JSON-based object as the state definition.

Code Snippets are available and can be plugged in as the state definition.

Custom states can be chained together with any of the other states to create your state machine definition. You will also need to provide any permissions that are required to the role that the State Machine uses.

The Retry and Catch fields are available for error handling. You can configure the Retry field by defining it in the JSON object or by adding it using the addRetry method. However, the Catch field cannot be configured by defining it in the JSON object, so it must be added using the addCatch method.

The following example uses the DynamoDB service integration to insert data into a DynamoDB table.

using Amazon.CDK.AWS.DynamoDB;


// create a table
var table = new Table(this, "montable", new TableProps {
    PartitionKey = new Attribute {
        Name = "id",
        Type = AttributeType.STRING
    }
});

var finalStatus = new Pass(this, "final step");

// States language JSON to put an item into DynamoDB
// snippet generated from https://docs.aws.amazon.com/step-functions/latest/dg/tutorial-code-snippet.html#tutorial-code-snippet-1
IDictionary<string, object> stateJson = new Dictionary<string, object> {
    { "Type", "Task" },
    { "Resource", "arn:aws:states:::dynamodb:putItem" },
    { "Parameters", new Dictionary<string, object> {
        { "TableName", table.TableName },
        { "Item", new Dictionary<string, IDictionary<string, string>> {
            { "id", new Dictionary<string, string> {
                { "S", "MyEntry" }
            } }
        } }
    } },
    { "ResultPath", null }
};

// custom state which represents a task to insert data into DynamoDB
var custom = new CustomState(this, "my custom task", new CustomStateProps {
    StateJson = stateJson
});

// catch errors with addCatch
var errorHandler = new Pass(this, "handle failure");
custom.AddCatch(errorHandler);

// retry the task if something goes wrong
custom.AddRetry(new RetryProps {
    Errors = new [] { Errors.ALL },
    Interval = Duration.Seconds(10),
    MaxAttempts = 5
});

var chain = Chain.Start(custom).Next(finalStatus);

var sm = new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(chain),
    Timeout = Duration.Seconds(30),
    Comment = "a super cool state machine"
});

// don't forget permissions. You need to assign them
table.GrantWriteData(sm);

Task Chaining

To make defining work flows as convenient (and readable in a top-to-bottom way) as writing regular programs, it is possible to chain most methods invocations. In particular, the .next() method can be repeated. The result of a series of .next() calls is called a Chain, and can be used when defining the jump targets of Choice.on or Parallel.branch:

var step1 = new Pass(this, "Step1");
var step2 = new Pass(this, "Step2");
var step3 = new Pass(this, "Step3");
var step4 = new Pass(this, "Step4");
var step5 = new Pass(this, "Step5");
var step6 = new Pass(this, "Step6");
var step7 = new Pass(this, "Step7");
var step8 = new Pass(this, "Step8");
var step9 = new Pass(this, "Step9");
var step10 = new Pass(this, "Step10");
var choice = new Choice(this, "Choice");
var condition1 = Condition.StringEquals("$.status", "SUCCESS");
var parallel = new Parallel(this, "Parallel");
var finish = new Pass(this, "Finish");

var definition = step1.Next(step2).Next(choice.When(condition1, step3.Next(step4).Next(step5)).Otherwise(step6).Afterwards()).Next(parallel.Branch(step7.Next(step8)).Branch(step9.Next(step10))).Next(finish);

new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition)
});

If you don't like the visual look of starting a chain directly off the first step, you can use Chain.start:

var step1 = new Pass(this, "Step1");
var step2 = new Pass(this, "Step2");
var step3 = new Pass(this, "Step3");

var definition = Chain.Start(step1).Next(step2).Next(step3);

Task Credentials

Tasks are executed using the State Machine's execution role. In some cases, e.g. cross-account access, an IAM role can be assumed by the State Machine's execution role to provide access to the resource. This can be achieved by providing the optional credentials property which allows using a fixed role or a json expression to resolve the role at runtime from the task's inputs.

using Amazon.CDK.AWS.Lambda;

Function submitLambda;
Role iamRole;


// use a fixed role for all task invocations
var role = TaskRole.FromRole(iamRole);
// or use a json expression to resolve the role at runtime based on task inputs
//const role = sfn.TaskRole.fromRoleArnJsonPath('$.RoleArn');

var submitJob = new LambdaInvoke(this, "Submit Job", new LambdaInvokeProps {
    LambdaFunction = submitLambda,
    OutputPath = "$.Payload",
    // use credentials
    Credentials = new Credentials { Role = role }
});

See the AWS documentation to learn more about AWS Step Functions support for accessing resources in other AWS accounts.

Service Integration Patterns

AWS Step functions integrate directly with other services, either through an optimised integration pattern, or through the AWS SDK. Therefore, it is possible to change the integrationPattern of services, to enable additional functionality of the said AWS Service:

using Amazon.CDK.AWS.Glue.Alpha;

Job submitGlue;


var submitJob = new GlueStartJobRun(this, "Submit Job", new GlueStartJobRunProps {
    GlueJobName = submitGlue.JobName,
    IntegrationPattern = IntegrationPattern.RUN_JOB
});

State Machine Fragments

It is possible to define reusable (or abstracted) mini-state machines by defining a construct that implements IChainable, which requires you to define two fields:

Since states will be named after their construct IDs, you may need to prefix the IDs of states if you plan to instantiate the same state machine fragment multiples times (otherwise all states in every instantiation would have the same name).

The class StateMachineFragment contains some helper functions (like prefixStates()) to make it easier for you to do this. If you define your state machine as a subclass of this, it will be convenient to use:

using Amazon.CDK;
using Constructs;
using Amazon.CDK.AWS.StepFunctions;

class MyJobProps
{
    public string JobFlavor { get; set; }
}

class MyJob : StateMachineFragment
{
    public State StartState { get; }
    public INextable[] EndStates { get; }

    public MyJob(Construct parent, string id, MyJobProps props) : base(parent, id)
    {

        var choice = new Choice(this, "Choice").When(Condition.StringEquals("$.branch", "left"), new Pass(this, "Left Branch")).When(Condition.StringEquals("$.branch", "right"), new Pass(this, "Right Branch"));

        // ...

        StartState = choice;
        EndStates = choice.Afterwards().EndStates;
    }
}

class MyStack : Stack
{
    public MyStack(Construct scope, string id) : base(scope, id)
    {
        // Do 3 different variants of MyJob in parallel
        var parallel = new Parallel(this, "All jobs").Branch(new MyJob(this, "Quick", new MyJobProps { JobFlavor = "quick" }).PrefixStates()).Branch(new MyJob(this, "Medium", new MyJobProps { JobFlavor = "medium" }).PrefixStates()).Branch(new MyJob(this, "Slow", new MyJobProps { JobFlavor = "slow" }).PrefixStates());

        new StateMachine(this, "MyStateMachine", new StateMachineProps {
            DefinitionBody = DefinitionBody.FromChainable(parallel)
        });
    }
}

A few utility functions are available to parse state machine fragments.

Activity

Activities represent work that is done on some non-Lambda worker pool. The Step Functions workflow will submit work to this Activity, and a worker pool that you run yourself, probably on EC2, will pull jobs from the Activity and submit the results of individual jobs back.

You need the ARN to do so, so if you use Activities be sure to pass the Activity ARN into your worker pool:

var activity = new Activity(this, "Activity");

// Read this CloudFormation Output from your application and use it to poll for work on
// the activity.
// Read this CloudFormation Output from your application and use it to poll for work on
// the activity.
new CfnOutput(this, "ActivityArn", new CfnOutputProps { Value = activity.ActivityArn });

Activity-Level Permissions

Granting IAM permissions to an activity can be achieved by calling the grant(principal, actions) API:

var activity = new Activity(this, "Activity");

var role = new Role(this, "Role", new RoleProps {
    AssumedBy = new ServicePrincipal("lambda.amazonaws.com")
});

activity.Grant(role, "states:SendTaskSuccess");

This will grant the IAM principal the specified actions onto the activity.

Metrics

Task object expose various metrics on the execution of that particular task. For example, to create an alarm on a particular task failing:

Task task;

new Alarm(this, "TaskAlarm", new AlarmProps {
    Metric = task.MetricFailed(),
    Threshold = 1,
    EvaluationPeriods = 1
});

There are also metrics on the complete state machine:

StateMachine stateMachine;

new Alarm(this, "StateMachineAlarm", new AlarmProps {
    Metric = stateMachine.MetricFailed(),
    Threshold = 1,
    EvaluationPeriods = 1
});

And there are metrics on the capacity of all state machines in your account:

new Alarm(this, "ThrottledAlarm", new AlarmProps {
    Metric = StateTransitionMetric.MetricThrottledEvents(),
    Threshold = 10,
    EvaluationPeriods = 2
});

Error names

Step Functions identifies errors in the Amazon States Language using case-sensitive strings, known as error names. The Amazon States Language defines a set of built-in strings that name well-known errors, all beginning with the States. prefix.

Logging

Enable logging to CloudWatch by passing a logging configuration with a destination LogGroup:

using Amazon.CDK.AWS.Logs;


var logGroup = new LogGroup(this, "MyLogGroup");

var definition = Chain.Start(new Pass(this, "Pass"));

new StateMachine(this, "MyStateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition),
    Logs = new LogOptions {
        Destination = logGroup,
        Level = LogLevel.ALL
    }
});

Encryption

You can encrypt your data using a customer managed key for AWS Step Functions state machines and activities. You can configure a symmetric AWS KMS key and data key reuse period when creating or updating a State Machine or when creating an Activity. The execution history and state machine definition will be encrypted with the key applied to the State Machine. Activity inputs will be encrypted with the key applied to the Activity.

Encrypting state machines

You can provide a symmetric KMS key to encrypt the state machine definition and execution history:

using Amazon.CDK.AWS.KMS;
using Amazon.CDK;


var kmsKey = new Key(this, "Key");
var stateMachine = new StateMachine(this, "StateMachineWithCMKEncryptionConfiguration", new StateMachineProps {
    StateMachineName = "StateMachineWithCMKEncryptionConfiguration",
    DefinitionBody = DefinitionBody.FromChainable(Chain.Start(new Pass(this, "Pass"))),
    StateMachineType = StateMachineType.STANDARD,
    EncryptionConfiguration = new CustomerManagedEncryptionConfiguration(kmsKey, Duration.Seconds(60))
});

Encrypting state machine logs in Cloud Watch Logs

If a state machine is encrypted with a customer managed key and has logging enabled, its decrypted execution history will be stored in CloudWatch Logs. If you want to encrypt the logs from the state machine using your own KMS key, you can do so by configuring the LogGroup associated with the state machine to use a KMS key.

using Amazon.CDK.AWS.KMS;
using Amazon.CDK;
using Amazon.CDK.AWS.Logs;


var stateMachineKmsKey = new Key(this, "StateMachine Key");
var logGroupKey = new Key(this, "LogGroup Key");

/*
  Required KMS key policy which allows the CloudWatchLogs service principal to encrypt the entire log group using the
  customer managed kms key. See: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/encrypt-log-data-kms.html#cmk-permissions
*/
logGroupKey.AddToResourcePolicy(new Aws_iam.PolicyStatement(new PolicyStatementProps {
    Resources = new [] { "*" },
    Actions = new [] { "kms:Encrypt*", "kms:Decrypt*", "kms:ReEncrypt*", "kms:GenerateDataKey*", "kms:Describe*" },
    Principals = new [] { new Aws_iam.ServicePrincipal($"logs.{cdk.Stack.of(this).region}.amazonaws.com") },
    Conditions = new Dictionary<string, object> {
        { "ArnEquals", new Dictionary<string, string> {
            { "kms:EncryptionContext:aws:logs:arn", Stack.Of(this).FormatArn(new ArnComponents {
                Service = "logs",
                Resource = "log-group",
                Sep = ":",
                ResourceName = "/aws/vendedlogs/states/MyLogGroup"
            }) }
        } }
    }
}));

// Create logGroup and provding encryptionKey which will be used to encrypt the log group
var logGroup = new LogGroup(this, "MyLogGroup", new LogGroupProps {
    LogGroupName = "/aws/vendedlogs/states/MyLogGroup",
    EncryptionKey = logGroupKey
});

// Create state machine with CustomerManagedEncryptionConfiguration
var stateMachine = new StateMachine(this, "StateMachineWithCMKWithCWLEncryption", new StateMachineProps {
    StateMachineName = "StateMachineWithCMKWithCWLEncryption",
    DefinitionBody = DefinitionBody.FromChainable(Chain.Start(new Pass(this, "PassState", new PassProps {
        Result = Result.FromString("Hello World")
    }))),
    StateMachineType = StateMachineType.STANDARD,
    EncryptionConfiguration = new CustomerManagedEncryptionConfiguration(stateMachineKmsKey),
    Logs = new LogOptions {
        Destination = logGroup,
        Level = LogLevel.ALL,
        IncludeExecutionData = true
    }
});

Encrypting activity inputs

When you provide a symmetric KMS key, all inputs from the Step Functions Activity will be encrypted using the provided KMS key:

using Amazon.CDK.AWS.KMS;
using Amazon.CDK;


var kmsKey = new Key(this, "Key");
var activity = new Activity(this, "ActivityWithCMKEncryptionConfiguration", new ActivityProps {
    ActivityName = "ActivityWithCMKEncryptionConfiguration",
    EncryptionConfiguration = new CustomerManagedEncryptionConfiguration(kmsKey, Duration.Seconds(75))
});

Changing Encryption

If you want to switch encryption from a customer provided key to a Step Functions owned key or vice-versa you must explicitly provide encryptionConfiguration?

Example: Switching from a customer managed key to a Step Functions owned key for StateMachine

Before

using Amazon.CDK.AWS.KMS;
using Amazon.CDK;


var kmsKey = new Key(this, "Key");
var stateMachine = new StateMachine(this, "StateMachine", new StateMachineProps {
    StateMachineName = "StateMachine",
    DefinitionBody = DefinitionBody.FromChainable(Chain.Start(new Pass(this, "Pass"))),
    StateMachineType = StateMachineType.STANDARD,
    EncryptionConfiguration = new CustomerManagedEncryptionConfiguration(kmsKey, Duration.Seconds(60))
});

After

var stateMachine = new StateMachine(this, "StateMachine", new StateMachineProps {
    StateMachineName = "StateMachine",
    DefinitionBody = DefinitionBody.FromChainable(Chain.Start(new Pass(this, "Pass"))),
    StateMachineType = StateMachineType.STANDARD,
    EncryptionConfiguration = new AwsOwnedEncryptionConfiguration()
});

X-Ray tracing

Enable X-Ray tracing for StateMachine:

var definition = Chain.Start(new Pass(this, "Pass"));

new StateMachine(this, "MyStateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition),
    TracingEnabled = true
});

See the AWS documentation to learn more about AWS Step Functions's X-Ray support.

State Machine Permission Grants

IAM roles, users, or groups which need to be able to work with a State Machine should be granted IAM permissions.

Any object that implements the IGrantable interface (has an associated principal) can be granted permissions by calling:

Start Execution Permission

Grant permission to start an execution of a state machine by calling the grantStartExecution() API.

IChainable definition;
var role = new Role(this, "Role", new RoleProps {
    AssumedBy = new ServicePrincipal("lambda.amazonaws.com")
});
var stateMachine = new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition)
});

// Give role permission to start execution of state machine
stateMachine.GrantStartExecution(role);

The following permission is provided to a service principal by the grantStartExecution() API:

Read Permissions

Grant read access to a state machine by calling the grantRead() API.

IChainable definition;
var role = new Role(this, "Role", new RoleProps {
    AssumedBy = new ServicePrincipal("lambda.amazonaws.com")
});
var stateMachine = new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition)
});

// Give role read access to state machine
stateMachine.GrantRead(role);

The following read permissions are provided to a service principal by the grantRead() API:

Task Response Permissions

Grant permission to allow task responses to a state machine by calling the grantTaskResponse() API:

IChainable definition;
var role = new Role(this, "Role", new RoleProps {
    AssumedBy = new ServicePrincipal("lambda.amazonaws.com")
});
var stateMachine = new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition)
});

// Give role task response permissions to the state machine
stateMachine.GrantTaskResponse(role);

The following read permissions are provided to a service principal by the grantRead() API:

Execution-level Permissions

Grant execution-level permissions to a state machine by calling the grantExecution() API:

IChainable definition;
var role = new Role(this, "Role", new RoleProps {
    AssumedBy = new ServicePrincipal("lambda.amazonaws.com")
});
var stateMachine = new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition)
});

// Give role permission to get execution history of ALL executions for the state machine
stateMachine.GrantExecution(role, "states:GetExecutionHistory");

Custom Permissions

You can add any set of permissions to a state machine by calling the grant() API.

IChainable definition;
var user = new User(this, "MyUser");
var stateMachine = new StateMachine(this, "StateMachine", new StateMachineProps {
    DefinitionBody = DefinitionBody.FromChainable(definition)
});

//give user permission to send task success to the state machine
stateMachine.Grant(user, "states:SendTaskSuccess");

Import

Any Step Functions state machine that has been created outside the stack can be imported into your CDK stack.

State machines can be imported by their ARN via the StateMachine.fromStateMachineArn() API. In addition, the StateMachine can be imported via the StateMachine.fromStateMachineName() method, as long as they are in the same account/region as the current construct.

var app = new App();
var stack = new Stack(app, "MyStack");
StateMachine.FromStateMachineArn(this, "ViaArnImportedStateMachine", "arn:aws:states:us-east-1:123456789012:stateMachine:StateMachine2E01A3A5-N5TJppzoevKQ");

StateMachine.FromStateMachineName(this, "ViaResourceNameImportedStateMachine", "StateMachine2E01A3A5-N5TJppzoevKQ");

Classes

Activity	Define a new Step Functions Activity.
ActivityProps	Properties for defining a new Step Functions Activity.
AfterwardsOptions	Options for selecting the choice paths.
AwsOwnedEncryptionConfiguration	Define a new AwsOwnedEncryptionConfiguration.
CatchProps	Error handler details.
CfnActivity	An activity is a task that you write in any programming language and host on any machine that has access to AWS Step Functions .
CfnActivity.EncryptionConfigurationProperty	Settings to configure server-side encryption for an activity.
CfnActivity.TagsEntryProperty	The TagsEntry property specifies tags to identify an activity.
CfnActivityProps	Properties for defining a CfnActivity.
CfnStateMachine	Provisions a state machine.
CfnStateMachine.CloudWatchLogsLogGroupProperty	Defines a CloudWatch log group.
CfnStateMachine.EncryptionConfigurationProperty	Settings to configure server-side encryption for a state machine.
CfnStateMachine.LogDestinationProperty	Defines a destination for LoggingConfiguration .
CfnStateMachine.LoggingConfigurationProperty	Defines what execution history events are logged and where they are logged.
CfnStateMachine.S3LocationProperty	Defines the S3 bucket location where a state machine definition is stored.
CfnStateMachine.TagsEntryProperty	The TagsEntry property specifies tags to identify a state machine.
CfnStateMachine.TracingConfigurationProperty	Selects whether or not the state machine's AWS X-Ray tracing is enabled.
CfnStateMachineAlias	Represents a state machine alias . An alias routes traffic to one or two versions of the same state machine.
CfnStateMachineAlias.DeploymentPreferenceProperty	Enables gradual state machine deployments.
CfnStateMachineAlias.RoutingConfigurationVersionProperty	The state machine version to which you want to route the execution traffic.
CfnStateMachineAliasProps	Properties for defining a CfnStateMachineAlias.
CfnStateMachineProps	Properties for defining a CfnStateMachine.
CfnStateMachineVersion	Represents a state machine version . A published version uses the latest state machine revision . A revision is an immutable, read-only snapshot of a state machine’s definition and configuration.
CfnStateMachineVersionProps	Properties for defining a CfnStateMachineVersion.
Chain	A collection of states to chain onto.
ChainDefinitionBody
Choice	Define a Choice in the state machine.
ChoiceProps	Properties for defining a Choice state.
ChoiceTransitionOptions	Options for Choice Transition.
Condition	A Condition for use in a Choice state branch.
Credentials	Specifies a target role assumed by the State Machine's execution role for invoking the task's resource.
CsvHeaderLocation	CSV header location options.
CsvHeaders	Configuration for CSV header options for a CSV Item Reader.
CustomerManagedEncryptionConfiguration	Define a new CustomerManagedEncryptionConfiguration.
CustomState	State defined by supplying Amazon States Language (ASL) in the state machine.
CustomStateProps	Properties for defining a custom state definition.
DefinitionBody
DefinitionConfig	Partial object from the StateMachine L1 construct properties containing definition information.
DistributedMap	Define a Distributed Mode Map state in the state machine.
DistributedMapProps	Properties for configuring a Distribute Map state.
EncryptionConfiguration	Base class for creating an EncryptionConfiguration for either state machines or activities.
Errors	Predefined error strings Error names in Amazon States Language - https://states-language.net/spec.html#appendix-a Error handling in Step Functions - https://docs.aws.amazon.com/step-functions/latest/dg/concepts-error-handling.html.
Fail	Define a Fail state in the state machine.
FailProps	Properties for defining a Fail state.
FieldUtils	Helper functions to work with structures containing fields.
FileDefinitionBody
FindStateOptions	Options for finding reachable states.
InputType	The type of task input.
IntegrationPattern	AWS Step Functions integrates with services directly in the Amazon States Language.
ItemBatcher	Configuration for processing a group of items in a single child workflow execution.
ItemBatcherProps	Interface for ItemBatcher configuration properties.
ItemReaderProps	Base interface for Item Reader configuration properties.
JitterType	Values allowed in the retrier JitterStrategy field.
JsonPath	Extract a field from the State Machine data or context that gets passed around between states.
LogLevel	Defines which category of execution history events are logged.
LogOptions	Defines what execution history events are logged and where they are logged.
Map	Define a Map state in the state machine.
MapBase	Define a Map state in the state machine.
MapBaseProps	Properties for defining a Map state.
MapProps	Properties for defining a Map state.
Parallel	Define a Parallel state in the state machine.
ParallelProps	Properties for defining a Parallel state.
Pass	Define a Pass in the state machine.
PassProps	Properties for defining a Pass state.
ProcessorConfig	Specifies the configuration for the processor Map state.
ProcessorMode	Mode of the Map workflow.
ProcessorType	Execution type for the Map workflow.
Result	The result of a Pass operation.
ResultWriter	Configuration for writing Distributed Map state results to S3.
ResultWriterProps	Interface for Result Writer configuration properties.
RetryProps	Retry details.
S3CsvItemReader	Item Reader configuration for iterating over items in a CSV file stored in S3.
S3CsvItemReaderProps	Properties for configuring an Item Reader that iterates over items in a CSV file in S3.
S3FileItemReaderProps	Base interface for Item Reader configuration properties the iterate over entries in a S3 file.
S3JsonItemReader	Item Reader configuration for iterating over items in a JSON array stored in a S3 file.
S3ManifestItemReader	Item Reader configuration for iterating over items in a S3 inventory manifest file stored in S3.
S3ObjectsItemReader	Item Reader configuration for iterating over objects in an S3 bucket.
S3ObjectsItemReaderProps	Properties for configuring an Item Reader that iterates over objects in an S3 bucket.
ServiceIntegrationPattern	Three ways to call an integrated service: Request Response, Run a Job and Wait for a Callback with Task Token.
SingleStateOptions	Options for creating a single state.
State	Base class for all other state classes.
StateGraph	A collection of connected states.
StateMachine	Define a StepFunctions State Machine.
StateMachineFragment	Base class for reusable state machine fragments.
StateMachineProps	Properties for defining a State Machine.
StateMachineType	Two types of state machines are available in AWS Step Functions: EXPRESS AND STANDARD.
StateProps	Properties shared by all states.
StateTransitionMetric	Metrics on the rate limiting performed on state machine execution.
StringDefinitionBody
Succeed	Define a Succeed state in the state machine.
SucceedProps	Properties for defining a Succeed state.
TaskInput	Type union for task classes that accept multiple types of payload.
TaskMetricsConfig	Task Metrics.
TaskRole	Role to be assumed by the State Machine's execution role for invoking a task's resource.
TaskStateBase	Define a Task state in the state machine.
TaskStateBaseProps	Props that are common to all tasks.
Timeout	Timeout for a task or heartbeat.
Wait	Define a Wait state in the state machine.
WaitProps	Properties for defining a Wait state.
WaitTime	Represents the Wait state which delays a state machine from continuing for a specified time.

Interfaces

CfnActivity.IEncryptionConfigurationProperty	Settings to configure server-side encryption for an activity.
CfnActivity.ITagsEntryProperty	The TagsEntry property specifies tags to identify an activity.
CfnStateMachine.ICloudWatchLogsLogGroupProperty	Defines a CloudWatch log group.
CfnStateMachine.IEncryptionConfigurationProperty	Settings to configure server-side encryption for a state machine.
CfnStateMachine.ILogDestinationProperty	Defines a destination for LoggingConfiguration .
CfnStateMachine.ILoggingConfigurationProperty	Defines what execution history events are logged and where they are logged.
CfnStateMachine.IS3LocationProperty	Defines the S3 bucket location where a state machine definition is stored.
CfnStateMachine.ITagsEntryProperty	The TagsEntry property specifies tags to identify a state machine.
CfnStateMachine.ITracingConfigurationProperty	Selects whether or not the state machine's AWS X-Ray tracing is enabled.
CfnStateMachineAlias.IDeploymentPreferenceProperty	Enables gradual state machine deployments.
CfnStateMachineAlias.IRoutingConfigurationVersionProperty	The state machine version to which you want to route the execution traffic.
IActivity	Represents a Step Functions Activity https://docs.aws.amazon.com/step-functions/latest/dg/concepts-activities.html.
IActivityProps	Properties for defining a new Step Functions Activity.
IAfterwardsOptions	Options for selecting the choice paths.
ICatchProps	Error handler details.
ICfnActivityProps	Properties for defining a CfnActivity.
ICfnStateMachineAliasProps	Properties for defining a CfnStateMachineAlias.
ICfnStateMachineProps	Properties for defining a CfnStateMachine.
ICfnStateMachineVersionProps	Properties for defining a CfnStateMachineVersion.
IChainable	Interface for objects that can be used in a Chain.
IChoiceProps	Properties for defining a Choice state.
IChoiceTransitionOptions	Options for Choice Transition.
ICredentials	Specifies a target role assumed by the State Machine's execution role for invoking the task's resource.
ICustomStateProps	Properties for defining a custom state definition.
IDefinitionConfig	Partial object from the StateMachine L1 construct properties containing definition information.
IDistributedMapProps	Properties for configuring a Distribute Map state.
IFailProps	Properties for defining a Fail state.
IFindStateOptions	Options for finding reachable states.
IItemBatcherProps	Interface for ItemBatcher configuration properties.
IItemReader	Base interface for Item Reader configurations.
IItemReaderProps	Base interface for Item Reader configuration properties.
ILogOptions	Defines what execution history events are logged and where they are logged.
IMapBaseProps	Properties for defining a Map state.
IMapProps	Properties for defining a Map state.
INextable	Interface for states that can have 'next' states.
IParallelProps	Properties for defining a Parallel state.
IPassProps	Properties for defining a Pass state.
IProcessorConfig	Specifies the configuration for the processor Map state.
IResultWriterProps	Interface for Result Writer configuration properties.
IRetryProps	Retry details.
IS3CsvItemReaderProps	Properties for configuring an Item Reader that iterates over items in a CSV file in S3.
IS3FileItemReaderProps	Base interface for Item Reader configuration properties the iterate over entries in a S3 file.
IS3ObjectsItemReaderProps	Properties for configuring an Item Reader that iterates over objects in an S3 bucket.
ISingleStateOptions	Options for creating a single state.
IStateMachine	A State Machine.
IStateMachineProps	Properties for defining a State Machine.
IStateProps	Properties shared by all states.
ISucceedProps	Properties for defining a Succeed state.
ITaskMetricsConfig	Task Metrics.
ITaskStateBaseProps	Props that are common to all tasks.
IWaitProps	Properties for defining a Wait state.