Learn the basics of Lambda with an AWS SDK (original) (raw)
The Cargo.toml with dependencies used in this scenario.
[package]
name = "lambda-code-examples"
version = "0.1.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
aws-config = { version = "1.0.1", features = ["behavior-version-latest"] }
aws-sdk-ec2 = { version = "1.3.0" }
aws-sdk-iam = { version = "1.3.0" }
aws-sdk-lambda = { version = "1.3.0" }
aws-sdk-s3 = { version = "1.4.0" }
aws-smithy-types = { version = "1.0.1" }
aws-types = { version = "1.0.1" }
clap = { version = "4.4", features = ["derive"] }
tokio = { version = "1.20.1", features = ["full"] }
tracing-subscriber = { version = "0.3.15", features = ["env-filter"] }
tracing = "0.1.37"
serde_json = "1.0.94"
anyhow = "1.0.71"
uuid = { version = "1.3.3", features = ["v4"] }
lambda_runtime = "0.8.0"
serde = "1.0.164"
A collection of utilities that streamline calling Lambda for this scenario. This file is src/ations.rs in the crate.
use anyhow::anyhow;
use aws_sdk_iam::operation::{create_role::CreateRoleError, delete_role::DeleteRoleOutput};
use aws_sdk_lambda::{
operation::{
delete_function::DeleteFunctionOutput, get_function::GetFunctionOutput,
invoke::InvokeOutput, list_functions::ListFunctionsOutput,
update_function_code::UpdateFunctionCodeOutput,
update_function_configuration::UpdateFunctionConfigurationOutput,
},
primitives::ByteStream,
types::{Environment, FunctionCode, LastUpdateStatus, State},
};
use aws_sdk_s3::{
error::ErrorMetadata,
operation::{delete_bucket::DeleteBucketOutput, delete_object::DeleteObjectOutput},
types::CreateBucketConfiguration,
};
use aws_smithy_types::Blob;
use serde::{ser::SerializeMap, Serialize};
use std::{fmt::Display, path::PathBuf, str::FromStr, time::Duration};
use tracing::{debug, info, warn};
/* Operation describes */
#[derive(Clone, Copy, Debug, Serialize)]
pub enum Operation {
#[serde(rename = "plus")]
Plus,
#[serde(rename = "minus")]
Minus,
#[serde(rename = "times")]
Times,
#[serde(rename = "divided-by")]
DividedBy,
}
impl FromStr for Operation {
type Err = anyhow::Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"plus" => Ok(Operation::Plus),
"minus" => Ok(Operation::Minus),
"times" => Ok(Operation::Times),
"divided-by" => Ok(Operation::DividedBy),
_ => Err(anyhow!("Unknown operation {s}")),
}
}
}
impl Display for Operation {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Operation::Plus => write!(f, "plus"),
Operation::Minus => write!(f, "minus"),
Operation::Times => write!(f, "times"),
Operation::DividedBy => write!(f, "divided-by"),
}
}
}
/**
* InvokeArgs will be serialized as JSON and sent to the AWS Lambda handler.
*/
#[derive(Debug)]
pub enum InvokeArgs {
Increment(i32),
Arithmetic(Operation, i32, i32),
}
impl Serialize for InvokeArgs {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
match self {
InvokeArgs::Increment(i) => serializer.serialize_i32(*i),
InvokeArgs::Arithmetic(o, i, j) => {
let mut map: S::SerializeMap = serializer.serialize_map(Some(3))?;
map.serialize_key(&"op".to_string())?;
map.serialize_value(&o.to_string())?;
map.serialize_key(&"i".to_string())?;
map.serialize_value(&i)?;
map.serialize_key(&"j".to_string())?;
map.serialize_value(&j)?;
map.end()
}
}
}
}
/** A policy document allowing Lambda to execute this function on the account's behalf. */
const ROLE_POLICY_DOCUMENT: &str = r#"{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": { "Service": "lambda.amazonaws.com" },
"Action": "sts:AssumeRole"
}
]
}"#;
/**
* A LambdaManager gathers all the resources necessary to run the Lambda example scenario.
* This includes instantiated aws_sdk clients and details of resource names.
*/
pub struct LambdaManager {
iam_client: aws_sdk_iam::Client,
lambda_client: aws_sdk_lambda::Client,
s3_client: aws_sdk_s3::Client,
lambda_name: String,
role_name: String,
bucket: String,
own_bucket: bool,
}
// These unit type structs provide nominal typing on top of String parameters for LambdaManager::new
pub struct LambdaName(pub String);
pub struct RoleName(pub String);
pub struct Bucket(pub String);
pub struct OwnBucket(pub bool);
impl LambdaManager {
pub fn new(
iam_client: aws_sdk_iam::Client,
lambda_client: aws_sdk_lambda::Client,
s3_client: aws_sdk_s3::Client,
lambda_name: LambdaName,
role_name: RoleName,
bucket: Bucket,
own_bucket: OwnBucket,
) -> Self {
Self {
iam_client,
lambda_client,
s3_client,
lambda_name: lambda_name.0,
role_name: role_name.0,
bucket: bucket.0,
own_bucket: own_bucket.0,
}
}
/**
* Load the AWS configuration from the environment.
* Look up lambda_name and bucket if none are given, or generate a random name if not present in the environment.
* If the bucket name is provided, the caller needs to have created the bucket.
* If the bucket name is generated, it will be created.
*/
pub async fn load_from_env(lambda_name: Option<String>, bucket: Option<String>) -> Self {
let sdk_config = aws_config::load_from_env().await;
let lambda_name = LambdaName(lambda_name.unwrap_or_else(|| {
std::env::var("LAMBDA_NAME").unwrap_or_else(|_| "rust_lambda_example".to_string())
}));
let role_name = RoleName(format!("{}_role", lambda_name.0));
let (bucket, own_bucket) =
match bucket {
Some(bucket) => (Bucket(bucket), false),
None => (
Bucket(std::env::var("LAMBDA_BUCKET").unwrap_or_else(|_| {
format!("rust-lambda-example-{}", uuid::Uuid::new_v4())
})),
true,
),
};
let s3_client = aws_sdk_s3::Client::new(&sdk_config);
if own_bucket {
info!("Creating bucket for demo: {}", bucket.0);
s3_client
.create_bucket()
.bucket(bucket.0.clone())
.create_bucket_configuration(
CreateBucketConfiguration::builder()
.location_constraint(aws_sdk_s3::types::BucketLocationConstraint::from(
sdk_config.region().unwrap().as_ref(),
))
.build(),
)
.send()
.await
.unwrap();
}
Self::new(
aws_sdk_iam::Client::new(&sdk_config),
aws_sdk_lambda::Client::new(&sdk_config),
s3_client,
lambda_name,
role_name,
bucket,
OwnBucket(own_bucket),
)
}
/**
* Upload function code from a path to a zip file.
* The zip file must have an AL2 Linux-compatible binary called `bootstrap`.
* The easiest way to create such a zip is to use `cargo lambda build --output-format Zip`.
*/
async fn prepare_function(
&self,
zip_file: PathBuf,
key: Option<String>,
) -> Result<FunctionCode, anyhow::Error> {
let body = ByteStream::from_path(zip_file).await?;
let key = key.unwrap_or_else(|| format!("{}_code", self.lambda_name));
info!("Uploading function code to s3://{}/{}", self.bucket, key);
let _ = self
.s3_client
.put_object()
.bucket(self.bucket.clone())
.key(key.clone())
.body(body)
.send()
.await?;
Ok(FunctionCode::builder()
.s3_bucket(self.bucket.clone())
.s3_key(key)
.build())
}
/**
* Create a function, uploading from a zip file.
*/
pub async fn create_function(&self, zip_file: PathBuf) -> Result<String, anyhow::Error> {
let code = self.prepare_function(zip_file, None).await?;
let key = code.s3_key().unwrap().to_string();
let role = self.create_role().await.map_err(|e| anyhow!(e))?;
info!("Created iam role, waiting 15s for it to become active");
tokio::time::sleep(Duration::from_secs(15)).await;
info!("Creating lambda function {}", self.lambda_name);
let _ = self
.lambda_client
.create_function()
.function_name(self.lambda_name.clone())
.code(code)
.role(role.arn())
.runtime(aws_sdk_lambda::types::Runtime::Providedal2)
.handler("_unused")
.send()
.await
.map_err(anyhow::Error::from)?;
self.wait_for_function_ready().await?;
self.lambda_client
.publish_version()
.function_name(self.lambda_name.clone())
.send()
.await?;
Ok(key)
}
/**
* Create an IAM execution role for the managed Lambda function.
* If the role already exists, use that instead.
*/
async fn create_role(&self) -> Result<aws_sdk_iam::types::Role, CreateRoleError> {
info!("Creating execution role for function");
let get_role = self
.iam_client
.get_role()
.role_name(self.role_name.clone())
.send()
.await;
if let Ok(get_role) = get_role {
if let Some(role) = get_role.role {
return Ok(role);
}
}
let create_role = self
.iam_client
.create_role()
.role_name(self.role_name.clone())
.assume_role_policy_document(ROLE_POLICY_DOCUMENT)
.send()
.await;
match create_role {
Ok(create_role) => match create_role.role {
Some(role) => Ok(role),
None => Err(CreateRoleError::generic(
ErrorMetadata::builder()
.message("CreateRole returned empty success")
.build(),
)),
},
Err(err) => Err(err.into_service_error()),
}
}
/**
* Poll `is_function_ready` with a 1-second delay. It returns when the function is ready or when there's an error checking the function's state.
*/
pub async fn wait_for_function_ready(&self) -> Result<(), anyhow::Error> {
info!("Waiting for function");
while !self.is_function_ready(None).await? {
info!("Function is not ready, sleeping 1s");
tokio::time::sleep(Duration::from_secs(1)).await;
}
Ok(())
}
/**
* Check if a Lambda function is ready to be invoked.
* A Lambda function is ready for this scenario when its state is active and its LastUpdateStatus is Successful.
* Additionally, if a sha256 is provided, the function must have that as its current code hash.
* Any missing properties or failed requests will be reported as an Err.
*/
async fn is_function_ready(
&self,
expected_code_sha256: Option<&str>,
) -> Result<bool, anyhow::Error> {
match self.get_function().await {
Ok(func) => {
if let Some(config) = func.configuration() {
if let Some(state) = config.state() {
info!(?state, "Checking if function is active");
if !matches!(state, State::Active) {
return Ok(false);
}
}
match config.last_update_status() {
Some(last_update_status) => {
info!(?last_update_status, "Checking if function is ready");
match last_update_status {
LastUpdateStatus::Successful => {
// continue
}
LastUpdateStatus::Failed | LastUpdateStatus::InProgress => {
return Ok(false);
}
unknown => {
warn!(
status_variant = unknown.as_str(),
"LastUpdateStatus unknown"
);
return Err(anyhow!(
"Unknown LastUpdateStatus, fn config is {config:?}"
));
}
}
}
None => {
warn!("Missing last update status");
return Ok(false);
}
};
if expected_code_sha256.is_none() {
return Ok(true);
}
if let Some(code_sha256) = config.code_sha256() {
return Ok(code_sha256 == expected_code_sha256.unwrap_or_default());
}
}
}
Err(e) => {
warn!(?e, "Could not get function while waiting");
}
}
Ok(false)
}
/** Get the Lambda function with this Manager's name. */
pub async fn get_function(&self) -> Result<GetFunctionOutput, anyhow::Error> {
info!("Getting lambda function");
self.lambda_client
.get_function()
.function_name(self.lambda_name.clone())
.send()
.await
.map_err(anyhow::Error::from)
}
/** List all Lambda functions in the current Region. */
pub async fn list_functions(&self) -> Result<ListFunctionsOutput, anyhow::Error> {
info!("Listing lambda functions");
self.lambda_client
.list_functions()
.send()
.await
.map_err(anyhow::Error::from)
}
/** Invoke the lambda function using calculator InvokeArgs. */
pub async fn invoke(&self, args: InvokeArgs) -> Result<InvokeOutput, anyhow::Error> {
info!(?args, "Invoking {}", self.lambda_name);
let payload = serde_json::to_string(&args)?;
debug!(?payload, "Sending payload");
self.lambda_client
.invoke()
.function_name(self.lambda_name.clone())
.payload(Blob::new(payload))
.send()
.await
.map_err(anyhow::Error::from)
}
/** Given a Path to a zip file, update the function's code and wait for the update to finish. */
pub async fn update_function_code(
&self,
zip_file: PathBuf,
key: String,
) -> Result<UpdateFunctionCodeOutput, anyhow::Error> {
let function_code = self.prepare_function(zip_file, Some(key)).await?;
info!("Updating code for {}", self.lambda_name);
let update = self
.lambda_client
.update_function_code()
.function_name(self.lambda_name.clone())
.s3_bucket(self.bucket.clone())
.s3_key(function_code.s3_key().unwrap().to_string())
.send()
.await
.map_err(anyhow::Error::from)?;
self.wait_for_function_ready().await?;
Ok(update)
}
/** Update the environment for a function. */
pub async fn update_function_configuration(
&self,
environment: Environment,
) -> Result<UpdateFunctionConfigurationOutput, anyhow::Error> {
info!(
?environment,
"Updating environment for {}", self.lambda_name
);
let updated = self
.lambda_client
.update_function_configuration()
.function_name(self.lambda_name.clone())
.environment(environment)
.send()
.await
.map_err(anyhow::Error::from)?;
self.wait_for_function_ready().await?;
Ok(updated)
}
/** Delete a function and its role, and if possible or necessary, its associated code object and bucket. */
pub async fn delete_function(
&self,
location: Option<String>,
) -> (
Result<DeleteFunctionOutput, anyhow::Error>,
Result<DeleteRoleOutput, anyhow::Error>,
Option<Result<DeleteObjectOutput, anyhow::Error>>,
) {
info!("Deleting lambda function {}", self.lambda_name);
let delete_function = self
.lambda_client
.delete_function()
.function_name(self.lambda_name.clone())
.send()
.await
.map_err(anyhow::Error::from);
info!("Deleting iam role {}", self.role_name);
let delete_role = self
.iam_client
.delete_role()
.role_name(self.role_name.clone())
.send()
.await
.map_err(anyhow::Error::from);
let delete_object: Option<Result<DeleteObjectOutput, anyhow::Error>> =
if let Some(location) = location {
info!("Deleting object {location}");
Some(
self.s3_client
.delete_object()
.bucket(self.bucket.clone())
.key(location)
.send()
.await
.map_err(anyhow::Error::from),
)
} else {
info!(?location, "Skipping delete object");
None
};
(delete_function, delete_role, delete_object)
}
pub async fn cleanup(
&self,
location: Option<String>,
) -> (
(
Result<DeleteFunctionOutput, anyhow::Error>,
Result<DeleteRoleOutput, anyhow::Error>,
Option<Result<DeleteObjectOutput, anyhow::Error>>,
),
Option<Result<DeleteBucketOutput, anyhow::Error>>,
) {
let delete_function = self.delete_function(location).await;
let delete_bucket = if self.own_bucket {
info!("Deleting bucket {}", self.bucket);
if delete_function.2.is_none() || delete_function.2.as_ref().unwrap().is_ok() {
Some(
self.s3_client
.delete_bucket()
.bucket(self.bucket.clone())
.send()
.await
.map_err(anyhow::Error::from),
)
} else {
None
}
} else {
info!("No bucket to clean up");
None
};
(delete_function, delete_bucket)
}
}
/**
* Testing occurs primarily as an integration test running the `scenario` bin successfully.
* Each action relies deeply on the internal workings and state of Amazon Simple Storage Service (Amazon S3), Lambda, and IAM working together.
* It is therefore infeasible to mock the clients to test the individual actions.
*/
#[cfg(test)]
mod test {
use super::{InvokeArgs, Operation};
use serde_json::json;
/** Make sure that the JSON output of serializing InvokeArgs is what's expected by the calculator. */
#[test]
fn test_serialize() {
assert_eq!(json!(InvokeArgs::Increment(5)), 5);
assert_eq!(
json!(InvokeArgs::Arithmetic(Operation::Plus, 5, 7)).to_string(),
r#"{"op":"plus","i":5,"j":7}"#.to_string(),
);
}
}
A binary to run the scenario from front to end, using command line flags to control some behavior. This file is src/bin/scenario.rs in the crate.
/*
## Service actions
Service actions wrap the SDK call, taking a client and any specific parameters necessary for the call.
* CreateFunction
* GetFunction
* ListFunctions
* Invoke
* UpdateFunctionCode
* UpdateFunctionConfiguration
* DeleteFunction
## Scenario
A scenario runs at a command prompt and prints output to the user on the result of each service action. A scenario can run in one of two ways: straight through, printing out progress as it goes, or as an interactive question/answer script.
## Getting started with functions
Use an SDK to manage AWS Lambda functions: create a function, invoke it, update its code, invoke it again, view its output and logs, and delete it.
This scenario uses two Lambda handlers:
_Note: Handlers don't use AWS SDK API calls._
The increment handler is straightforward:
1. It accepts a number, increments it, and returns the new value.
2. It performs simple logging of the result.
The arithmetic handler is more complex:
1. It accepts a set of actions ['plus', 'minus', 'times', 'divided-by'] and two numbers, and returns the result of the calculation.
2. It uses an environment variable to control log level (such as DEBUG, INFO, WARNING, ERROR).
It logs a few things at different levels, such as:
* DEBUG: Full event data.
* INFO: The calculation result.
* WARN~ING~: When a divide by zero error occurs.
* This will be the typical `RUST_LOG` variable.
The steps of the scenario are:
1. Create an AWS Identity and Access Management (IAM) role that meets the following requirements:
* Has an assume_role policy that grants 'lambda.amazonaws.com' the 'sts:AssumeRole' action.
* Attaches the 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole' managed role.
* _You must wait for ~10 seconds after the role is created before you can use it!_
2. Create a function (CreateFunction) for the increment handler by packaging it as a zip and doing one of the following:
* Adding it with CreateFunction Code.ZipFile.
* --or--
* Uploading it to Amazon Simple Storage Service (Amazon S3) and adding it with CreateFunction Code.S3Bucket/S3Key.
* _Note: Zipping the file does not have to be done in code._
* If you have a waiter, use it to wait until the function is active. Otherwise, call GetFunction until State is Active.
3. Invoke the function with a number and print the result.
4. Update the function (UpdateFunctionCode) to the arithmetic handler by packaging it as a zip and doing one of the following:
* Adding it with UpdateFunctionCode ZipFile.
* --or--
* Uploading it to Amazon S3 and adding it with UpdateFunctionCode S3Bucket/S3Key.
5. Call GetFunction until Configuration.LastUpdateStatus is 'Successful' (or 'Failed').
6. Update the environment variable by calling UpdateFunctionConfiguration and pass it a log level, such as:
* Environment={'Variables': {'RUST_LOG': 'TRACE'}}
7. Invoke the function with an action from the list and a couple of values. Include LogType='Tail' to get logs in the result. Print the result of the calculation and the log.
8. [Optional] Invoke the function to provoke a divide-by-zero error and show the log result.
9. List all functions for the account, using pagination (ListFunctions).
10. Delete the function (DeleteFunction).
11. Delete the role.
Each step should use the function created in Service Actions to abstract calling the SDK.
*/
use aws_sdk_lambda::{operation::invoke::InvokeOutput, types::Environment};
use clap::Parser;
use std::{collections::HashMap, path::PathBuf};
use tracing::{debug, info, warn};
use tracing_subscriber::EnvFilter;
use lambda_code_examples::actions::{
InvokeArgs::{Arithmetic, Increment},
LambdaManager, Operation,
};
#[derive(Debug, Parser)]
pub struct Opt {
/// The AWS Region.
#[structopt(short, long)]
pub region: Option<String>,
// The bucket to use for the FunctionCode.
#[structopt(short, long)]
pub bucket: Option<String>,
// The name of the Lambda function.
#[structopt(short, long)]
pub lambda_name: Option<String>,
// The number to increment.
#[structopt(short, long, default_value = "12")]
pub inc: i32,
// The left operand.
#[structopt(long, default_value = "19")]
pub num_a: i32,
// The right operand.
#[structopt(long, default_value = "23")]
pub num_b: i32,
// The arithmetic operation.
#[structopt(short, long, default_value = "plus")]
pub operation: Operation,
#[structopt(long)]
pub cleanup: Option<bool>,
#[structopt(long)]
pub no_cleanup: Option<bool>,
}
fn code_path(lambda: &str) -> PathBuf {
PathBuf::from(format!("../target/lambda/{lambda}/bootstrap.zip"))
}
fn log_invoke_output(invoke: &InvokeOutput, message: &str) {
if let Some(payload) = invoke.payload().cloned() {
let payload = String::from_utf8(payload.into_inner());
info!(?payload, message);
} else {
info!("Could not extract payload")
}
if let Some(logs) = invoke.log_result() {
debug!(?logs, "Invoked function logs")
} else {
debug!("Invoked function had no logs")
}
}
async fn main_block(
opt: &Opt,
manager: &LambdaManager,
code_location: String,
) -> Result<(), anyhow::Error> {
let invoke = manager.invoke(Increment(opt.inc)).await?;
log_invoke_output(&invoke, "Invoked function configured as increment");
let update_code = manager
.update_function_code(code_path("arithmetic"), code_location.clone())
.await?;
let code_sha256 = update_code.code_sha256().unwrap_or("Unknown SHA");
info!(?code_sha256, "Updated function code with arithmetic.zip");
let arithmetic_args = Arithmetic(opt.operation, opt.num_a, opt.num_b);
let invoke = manager.invoke(arithmetic_args).await?;
log_invoke_output(&invoke, "Invoked function configured as arithmetic");
let update = manager
.update_function_configuration(
Environment::builder()
.set_variables(Some(HashMap::from([(
"RUST_LOG".to_string(),
"trace".to_string(),
)])))
.build(),
)
.await?;
let updated_environment = update.environment();
info!(?updated_environment, "Updated function configuration");
let invoke = manager
.invoke(Arithmetic(opt.operation, opt.num_a, opt.num_b))
.await?;
log_invoke_output(
&invoke,
"Invoked function configured as arithmetic with increased logging",
);
let invoke = manager
.invoke(Arithmetic(Operation::DividedBy, opt.num_a, 0))
.await?;
log_invoke_output(
&invoke,
"Invoked function configured as arithmetic with divide by zero",
);
Ok::<(), anyhow::Error>(())
}
#[tokio::main]
async fn main() {
tracing_subscriber::fmt()
.without_time()
.with_file(true)
.with_line_number(true)
.with_env_filter(EnvFilter::from_default_env())
.init();
let opt = Opt::parse();
let manager = LambdaManager::load_from_env(opt.lambda_name.clone(), opt.bucket.clone()).await;
let key = match manager.create_function(code_path("increment")).await {
Ok(init) => {
info!(?init, "Created function, initially with increment.zip");
let run_block = main_block(&opt, &manager, init.clone()).await;
info!(?run_block, "Finished running example, cleaning up");
Some(init)
}
Err(err) => {
warn!(?err, "Error happened when initializing function");
None
}
};
if Some(false) == opt.cleanup || Some(true) == opt.no_cleanup {
info!("Skipping cleanup")
} else {
let delete = manager.cleanup(key).await;
info!(?delete, "Deleted function & cleaned up resources");
}
}
- For API details, see the following topics in AWS SDK for Rust API reference.