diff --git a/README.md b/README.md index 1465a9cc..898cc5b1 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,7 @@ It contains the following modules: * [SpringBoot](/springboot): showcases SpringBoot autoconfig integration. * [SpringBoot Basic](/springboot-basic): Minimal sample showing SpringBoot autoconfig integration without any extra external dependencies. * [Spring AI](/springai): demonstrates the Temporal Spring AI integration — durable AI agents with chat models, tools, MCP servers, vector stores, and embeddings. +* [Lambda Worker](/lambda-worker): demonstrates running a Temporal Java Worker inside AWS Lambda. ## Learn more about Temporal and Java SDK @@ -115,6 +116,8 @@ See the README.md file in each main sample directory for cut/paste Gradle comman - [**Environment Configuration**](/core/src/main/java/io/temporal/samples/envconfig): Load client configuration from TOML files with programmatic overrides. +- [**Lambda Worker**](/lambda-worker): Demonstrates running a Temporal Java Worker inside AWS Lambda with Worker Deployment Versioning. + #### API demonstrations - [**Workflow Streams**](/core/src/main/java/io/temporal/samples/workflowstreams): Demonstrates a durable publish/subscribe log hosted inside a workflow, using the experimental `temporal-workflowstreams` module. diff --git a/lambda-worker/.gitignore b/lambda-worker/.gitignore new file mode 100644 index 00000000..69b7c323 --- /dev/null +++ b/lambda-worker/.gitignore @@ -0,0 +1,2 @@ +temporal.toml +otel-collector-config.yaml diff --git a/lambda-worker/README.md b/lambda-worker/README.md new file mode 100644 index 00000000..dd9e599b --- /dev/null +++ b/lambda-worker/README.md @@ -0,0 +1,342 @@ +# Lambda Worker + +This sample demonstrates a Temporal Java Worker running inside an AWS Lambda function. +It registers a simple greeting Workflow and Activity, configures Worker Deployment +Versioning, and includes helper scripts for packaging the Lambda and configuring Temporal +Cloud invocation. + +The deployable Worker and local Workflow starter are separate Gradle projects, so +starter-only code and dependencies are not included in the Lambda artifact. + +It uses the same published Temporal Java SDK version as the other samples in this repository. + +## Prerequisites + +- Java 17+ +- AWS CLI configured with permissions to create Lambda functions, IAM roles, and + CloudFormation stacks +- An AWS-hosted Temporal Cloud namespace with Serverless Workers enabled, or a + [self-hosted Temporal Service](https://docs.temporal.io/production-deployment/worker-deployments/serverless-workers/self-hosted-setup) + version 1.31.0 or later with the AWS Lambda Worker Controller setup completed +- A Temporal Cloud API key (if using Temporal Cloud). This walkthrough deploys it as a + Lambda environment variable because these are development-only secrets. + +## Layout + +- `worker/` contains the Lambda handler, Workflow, Activity, and deployable Worker project. +- `starter/` contains the local Workflow starter project. +- `deploy/` contains the AWS deployment scripts and CloudFormation template. +- `temporal.template.toml` is a Temporal connection configuration template. +- `otel-collector-config.template.yaml` configures the ADOT collector packaged with the + Lambda Worker. + +## Build + +```bash +./gradlew :lambda-worker:worker:test +./gradlew :lambda-worker:worker:shadowJar +``` + +The `shadowJar` task packages `otel-collector-config.template.yaml` at the root of the Lambda +artifact as `otel-collector-config.yaml`. + +The Lambda handler string is: + +```text +io.temporal.samples.lambdaworker.LambdaFunction::handleRequest +``` + +## Configure Environment + +Set AWS, Temporal, and sample names first. Use unique values if you share the account or +namespace with other developers. The connection values below are for Temporal Cloud. For a +self-hosted Service, use its frontend address, Namespace, and TLS or authentication settings; +leave `TEMPORAL_API_KEY` unset if the Service does not require one. + +```bash +export AWS_PROFILE= +export AWS_REGION=us-west-2 +export AWS_DEFAULT_REGION="$AWS_REGION" +export AWS_ACCOUNT_ID="$(aws sts get-caller-identity --query Account --output text)" + +export TEMPORAL_ADDRESS=..tmprl.cloud:7233 +export TEMPORAL_NAMESPACE=. +export TEMPORAL_API_KEY= +export TEMPORAL_TLS=true + +export FUNCTION_NAME=my-temporal-java-worker +export EXECUTION_ROLE_NAME="${FUNCTION_NAME}-exec" +export STACK_NAME="${FUNCTION_NAME}-invoke" +export EXTERNAL_ID="${FUNCTION_NAME}-external-id" + +export DEPLOYMENT_NAME=my-app +export BUILD_ID=build-1 +export TASK_QUEUE=serverless-task-queue-java +export WORKFLOW_PREFIX=serverless-workflow-id-java +``` + +The Lambda worker reads these environment variables: + +```bash +TEMPORAL_ADDRESS +TEMPORAL_NAMESPACE +TEMPORAL_API_KEY +TEMPORAL_TASK_QUEUE +TEMPORAL_LAMBDA_DEPLOYMENT_NAME +TEMPORAL_LAMBDA_BUILD_ID +``` + +The ADOT collector extension reads +`OPENTELEMETRY_COLLECTOR_CONFIG_URI=/var/task/otel-collector-config.yaml`. The Java Lambda Worker +uses `OtelLambdaWorkerConfigurationHelper` to send Temporal traces and metrics to the collector +over OTLP. The collector exports traces to AWS X-Ray and metrics to the +`TemporalWorkerMetrics` CloudWatch namespace. + +The local starter also reads `TEMPORAL_TASK_QUEUE` and +`TEMPORAL_LAMBDA_WORKFLOW_ID_PREFIX`. + +You can also copy `lambda-worker/temporal.template.toml` to +`lambda-worker/temporal.toml`, fill in the connection details, and set +`TEMPORAL_CONFIG_FILE=lambda-worker/temporal.toml`. The `temporal.toml` file is ignored by Git. + +`TEMPORAL_TASK_QUEUE`, `TEMPORAL_LAMBDA_DEPLOYMENT_NAME`, +`TEMPORAL_LAMBDA_BUILD_ID`, and `TEMPORAL_LAMBDA_WORKFLOW_ID_PREFIX` are optional. The +values above are the sample defaults. + +## Deploy Lambda + +Create the Lambda execution role: + +```bash +cat > /tmp/temporal-lambda-trust-policy.json <<'JSON' +{ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": { + "Service": "lambda.amazonaws.com" + }, + "Action": "sts:AssumeRole" + } + ] +} +JSON + +aws iam create-role \ + --role-name "$EXECUTION_ROLE_NAME" \ + --assume-role-policy-document file:///tmp/temporal-lambda-trust-policy.json \ + --query 'Role.Arn' \ + --output text + +aws iam attach-role-policy \ + --role-name "$EXECUTION_ROLE_NAME" \ + --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole + +export EXECUTION_ROLE_ARN="$( + aws iam get-role \ + --role-name "$EXECUTION_ROLE_NAME" \ + --query 'Role.Arn' \ + --output text +)" +``` + +Build the deployment jar and create the Java 17 Lambda function. The jar is small enough +for direct upload in this sample; use S3 if your local artifact grows beyond Lambda's +direct upload limit. + +```bash +./gradlew :lambda-worker:worker:shadowJar + +aws lambda create-function \ + --function-name "$FUNCTION_NAME" \ + --runtime java17 \ + --handler io.temporal.samples.lambdaworker.LambdaFunction::handleRequest \ + --role "$EXECUTION_ROLE_ARN" \ + --zip-file fileb://lambda-worker/worker/build/libs/lambda-worker-1.0.0-all.jar \ + --environment "Variables={TEMPORAL_ADDRESS=$TEMPORAL_ADDRESS,TEMPORAL_NAMESPACE=$TEMPORAL_NAMESPACE,TEMPORAL_API_KEY=$TEMPORAL_API_KEY,TEMPORAL_TASK_QUEUE=$TASK_QUEUE,TEMPORAL_LAMBDA_DEPLOYMENT_NAME=$DEPLOYMENT_NAME,TEMPORAL_LAMBDA_BUILD_ID=$BUILD_ID,OPENTELEMETRY_COLLECTOR_CONFIG_URI=/var/task/otel-collector-config.yaml}" \ + --timeout 90 \ + --memory-size 1024 \ + --query 'FunctionArn' \ + --output text + +aws lambda wait function-active --function-name "$FUNCTION_NAME" + +./lambda-worker/deploy/enable-telemetry.sh \ + "$EXECUTION_ROLE_NAME" \ + "$FUNCTION_NAME" \ + "$AWS_REGION" \ + "$AWS_ACCOUNT_ID" + +aws lambda wait function-updated --function-name "$FUNCTION_NAME" + +export FUNCTION_BASE_ARN="$( + aws lambda get-function \ + --function-name "$FUNCTION_NAME" \ + --query 'Configuration.FunctionArn' \ + --output text +)" + +export FUNCTION_VERSION_ARN="$( + aws lambda publish-version \ + --function-name "$FUNCTION_NAME" \ + --description "Build ID $BUILD_ID" \ + --query 'FunctionArn' \ + --output text +)" +``` + +`enable-telemetry.sh` grants the execution role permission to send traces and EMF logs, enables +active tracing, and attaches AWS's collector-only ADOT Lambda layer. Its default layer ARN is for +the sample's `x86_64` architecture in standard AWS regions. Set `ADOT_COLLECTOR_LAYER_ARN` before +running the script to use another compatible regional layer. + +## Configure Invocation + +For Temporal Cloud, create the IAM role that Temporal Cloud assumes to invoke the Lambda. The +wildcard suffix authorizes invocation of every immutable version published for this function: + +```bash +./lambda-worker/deploy/mk-iam-role.sh \ + "$STACK_NAME" \ + "$EXTERNAL_ID" \ + "${FUNCTION_BASE_ARN}:*" + +aws cloudformation wait stack-create-complete --stack-name "$STACK_NAME" + +export INVOCATION_ROLE_ARN="$( + aws cloudformation describe-stacks \ + --stack-name "$STACK_NAME" \ + --query "Stacks[0].Outputs[?OutputKey=='RoleARN'].OutputValue | [0]" \ + --output text +)" +``` + +The included CloudFormation template trusts Temporal Cloud's AWS identities and must not be +used for a self-hosted Temporal Service. For self-hosted deployments, complete the +[self-hosted Serverless Workers setup](https://docs.temporal.io/production-deployment/worker-deployments/serverless-workers/self-hosted-setup), +then set `INVOCATION_ROLE_ARN` to the role created by that process. + +## Create Worker Deployment Version + +Create and route the Worker Deployment Version. The Temporal CLI can connect to either Temporal +Cloud or a self-hosted Service using the connection configuration above. + +```bash +temporal worker deployment create --name "$DEPLOYMENT_NAME" + +temporal worker deployment create-version \ + --deployment-name "$DEPLOYMENT_NAME" \ + --build-id "$BUILD_ID" \ + --aws-lambda-function-arn "$FUNCTION_VERSION_ARN" \ + --aws-lambda-assume-role-arn "$INVOCATION_ROLE_ARN" \ + --aws-lambda-assume-role-external-id "$EXTERNAL_ID" + +temporal worker deployment set-current-version \ + --deployment-name "$DEPLOYMENT_NAME" \ + --build-id "$BUILD_ID" \ + --allow-no-pollers \ + --yes +``` + +An async Lambda smoke test returns immediately and should produce worker startup logs: + +```bash +aws lambda invoke \ + --function-name "$FUNCTION_VERSION_ARN" \ + --invocation-type Event \ + --cli-binary-format raw-in-base64-out \ + --payload '{}' \ + /tmp/lambda-worker-response.json \ + --query 'StatusCode' \ + --output text +``` + +A synchronous invoke can run until the Lambda worker exits near the function timeout. If +you want to wait for that path, set the AWS CLI read timeout higher than the function +timeout. + +## Deploy an Updated Version + +Each Temporal Build ID should point to an immutable Lambda function version. To deploy an +update, choose a new Build ID, update the Lambda environment, upload the new code, and publish a +new Lambda version: + +```bash +export BUILD_ID=build-2 + +aws lambda update-function-configuration \ + --function-name "$FUNCTION_NAME" \ + --environment "Variables={TEMPORAL_ADDRESS=$TEMPORAL_ADDRESS,TEMPORAL_NAMESPACE=$TEMPORAL_NAMESPACE,TEMPORAL_API_KEY=$TEMPORAL_API_KEY,TEMPORAL_TASK_QUEUE=$TASK_QUEUE,TEMPORAL_LAMBDA_DEPLOYMENT_NAME=$DEPLOYMENT_NAME,TEMPORAL_LAMBDA_BUILD_ID=$BUILD_ID,OPENTELEMETRY_COLLECTOR_CONFIG_URI=/var/task/otel-collector-config.yaml}" \ + --query 'FunctionArn' \ + --output text + +aws lambda wait function-updated --function-name "$FUNCTION_NAME" + +./lambda-worker/deploy/deploy-lambda.sh "$FUNCTION_NAME" + +aws lambda wait function-updated --function-name "$FUNCTION_NAME" + +export FUNCTION_VERSION_ARN="$( + aws lambda publish-version \ + --function-name "$FUNCTION_NAME" \ + --description "Build ID $BUILD_ID" \ + --query 'FunctionArn' \ + --output text +)" +``` + +If direct upload is too large, set `LAMBDA_CODE_S3_BUCKET` when running `deploy-lambda.sh`: + +```bash +LAMBDA_CODE_S3_BUCKET= ./lambda-worker/deploy/deploy-lambda.sh "$FUNCTION_NAME" +``` + +Create the new Worker Deployment Version and make it current using the commands in +[Create Worker Deployment Version](#create-worker-deployment-version), omitting the +`temporal worker deployment create` command because the deployment already exists. Existing +Worker Deployment Versions continue to reference their original Lambda versions. + +## Start Workflow + +After the Worker Deployment Version is current, start the sample Workflow: + +```bash +export TEMPORAL_TASK_QUEUE="$TASK_QUEUE" +export TEMPORAL_LAMBDA_WORKFLOW_ID_PREFIX="$WORKFLOW_PREFIX" + +./gradlew -q :lambda-worker:starter:execute +``` + +The starter only creates a Workflow Execution. It does not start a local Worker. The +important value is `TEMPORAL_TASK_QUEUE`; it must match the task queue configured on the +Lambda function. + +## Verify OpenTelemetry + +After invoking the Lambda or completing a Workflow, inspect the function logs for ADOT collector +startup and export messages: + +```bash +aws logs tail "/aws/lambda/$FUNCTION_NAME" --since 10m +``` + +Temporal SDK metrics should also appear in the `TemporalWorkerMetrics` CloudWatch namespace: + +```bash +aws cloudwatch list-metrics \ + --namespace TemporalWorkerMetrics \ + --query 'Metrics[].MetricName' \ + --output text +``` + +Temporal tracing spans are exported to AWS X-Ray and can be inspected in the X-Ray trace view. + +## Local SDK Development + +For local development of the Workflow and Activity logic, run the unit tests. They use +`TestWorkflowRule` and do not require AWS or a running Temporal Service. + +```bash +./gradlew :lambda-worker:worker:test +``` diff --git a/lambda-worker/deploy/deploy-lambda.sh b/lambda-worker/deploy/deploy-lambda.sh new file mode 100755 index 00000000..7341193e --- /dev/null +++ b/lambda-worker/deploy/deploy-lambda.sh @@ -0,0 +1,41 @@ +#!/bin/bash +set -euo pipefail + +FUNCTION_NAME="${1:?Usage: deploy-lambda.sh }" +MAX_DIRECT_UPLOAD_BYTES=50000000 + +REPOSITORY_DIR="$(cd "$(dirname "$0")/../.." && pwd)" +cd "$REPOSITORY_DIR" +./gradlew :lambda-worker:worker:shadowJar + +JAR_FILE="$(find lambda-worker/worker/build/libs -name 'lambda-worker-*-all.jar' | head -n 1)" + +if stat -f%z "$JAR_FILE" >/dev/null 2>&1; then + JAR_SIZE="$(stat -f%z "$JAR_FILE")" +else + JAR_SIZE="$(stat -c%s "$JAR_FILE")" +fi + +if [[ -n "${LAMBDA_CODE_S3_BUCKET:-}" ]]; then + S3_KEY="${LAMBDA_CODE_S3_KEY:-lambda-worker/$(basename "$JAR_FILE")}" + aws s3 cp "$JAR_FILE" "s3://$LAMBDA_CODE_S3_BUCKET/$S3_KEY" + aws lambda update-function-code \ + --function-name "$FUNCTION_NAME" \ + --s3-bucket "$LAMBDA_CODE_S3_BUCKET" \ + --s3-key "$S3_KEY" \ + --query 'FunctionArn' \ + --output text + exit 0 +fi + +if (( JAR_SIZE > MAX_DIRECT_UPLOAD_BYTES )); then + echo "Artifact is ${JAR_SIZE} bytes, which is too large for direct Lambda upload." >&2 + echo "Set LAMBDA_CODE_S3_BUCKET and rerun to upload through S3." >&2 + exit 1 +fi + +aws lambda update-function-code \ + --function-name "$FUNCTION_NAME" \ + --zip-file "fileb://$JAR_FILE" \ + --query 'FunctionArn' \ + --output text diff --git a/lambda-worker/deploy/enable-telemetry.sh b/lambda-worker/deploy/enable-telemetry.sh new file mode 100755 index 00000000..209ebf78 --- /dev/null +++ b/lambda-worker/deploy/enable-telemetry.sh @@ -0,0 +1,41 @@ +#!/bin/bash +set -euo pipefail + +ROLE_NAME="${1:?Usage: enable-telemetry.sh }" +FUNCTION_NAME="${2:?Usage: enable-telemetry.sh }" +REGION="${3:?Usage: enable-telemetry.sh }" +ACCOUNT_ID="${4:?Usage: enable-telemetry.sh }" +ADOT_COLLECTOR_LAYER_ARN="${ADOT_COLLECTOR_LAYER_ARN:-arn:aws:lambda:${REGION}:901920570463:layer:aws-otel-collector-amd64-ver-0-117-0:1}" + +aws iam put-role-policy \ + --role-name "$ROLE_NAME" \ + --policy-name ADOT-Telemetry-Permissions \ + --policy-document "{ + \"Version\": \"2012-10-17\", + \"Statement\": [ + { + \"Effect\": \"Allow\", + \"Action\": [ + \"logs:CreateLogGroup\", + \"logs:CreateLogStream\", + \"logs:PutLogEvents\" + ], + \"Resource\": \"arn:aws:logs:${REGION}:${ACCOUNT_ID}:log-group:/aws/lambda/${FUNCTION_NAME}:*\" + }, + { + \"Effect\": \"Allow\", + \"Action\": [ + \"xray:PutTraceSegments\", + \"xray:PutTelemetryRecords\" + ], + \"Resource\": \"*\" + } + ] + }" + +aws lambda update-function-configuration \ + --function-name "$FUNCTION_NAME" \ + --layers "$ADOT_COLLECTOR_LAYER_ARN" \ + --tracing-config Mode=Active \ + --query 'FunctionArn' \ + --output text diff --git a/lambda-worker/deploy/mk-iam-role.sh b/lambda-worker/deploy/mk-iam-role.sh new file mode 100755 index 00000000..d0b3d535 --- /dev/null +++ b/lambda-worker/deploy/mk-iam-role.sh @@ -0,0 +1,15 @@ +#!/bin/bash +set -euo pipefail + +STACK_NAME="${1:?Usage: mk-iam-role.sh }" +EXTERNAL_ID="${2:?Usage: mk-iam-role.sh }" +LAMBDA_ARN_PATTERN="${3:?Usage: mk-iam-role.sh }" +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" + +aws cloudformation create-stack \ + --stack-name "$STACK_NAME" \ + --template-body "file://${SCRIPT_DIR}/temporal-cloud-lambda-invoke-role.yaml" \ + --parameters \ + ParameterKey=AssumeRoleExternalId,ParameterValue="$EXTERNAL_ID" \ + ParameterKey=LambdaFunctionARNs,ParameterValue="\"$LAMBDA_ARN_PATTERN\"" \ + --capabilities CAPABILITY_IAM diff --git a/lambda-worker/deploy/temporal-cloud-lambda-invoke-role.yaml b/lambda-worker/deploy/temporal-cloud-lambda-invoke-role.yaml new file mode 100644 index 00000000..e048f446 --- /dev/null +++ b/lambda-worker/deploy/temporal-cloud-lambda-invoke-role.yaml @@ -0,0 +1,87 @@ +# This trust policy is for Temporal Cloud. Self-hosted Temporal Services must trust their +# own AWS identity; see the self-hosted Serverless Workers setup documentation. +AWSTemplateFormatVersion: "2010-09-09" +Description: Creates an IAM role that Temporal Cloud can assume to invoke Lambda functions for Serverless Workers. + +Parameters: + AssumeRoleExternalId: + Type: String + Description: A string you choose. Use the same value when creating the Worker Deployment Version. + AllowedPattern: "[a-zA-Z0-9_+=,.@-]*" + MinLength: 5 + MaxLength: 45 + + LambdaFunctionARNs: + Type: CommaDelimitedList + Description: Comma-separated list of Lambda function ARNs to invoke. + +Metadata: + AWS::CloudFormation::Interface: + ParameterGroups: + - Label: + default: "Temporal Cloud Configuration" + Parameters: + - AssumeRoleExternalId + - Label: + default: "Lambda Configuration" + Parameters: + - LambdaFunctionARNs + ParameterLabels: + AssumeRoleExternalId: + default: "External ID" + LambdaFunctionARNs: + default: "Lambda Function ARNs" +Resources: + TemporalCloudServerlessWorker: + Type: AWS::IAM::Role + Properties: + AssumeRolePolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Principal: + AWS: + [ + arn:aws:iam::902542641901:role/wci-lambda-invoke, + arn:aws:iam::160190466495:role/wci-lambda-invoke, + arn:aws:iam::819232936619:role/wci-lambda-invoke, + arn:aws:iam::829909441867:role/wci-lambda-invoke, + arn:aws:iam::354116250941:role/wci-lambda-invoke, + ] + Action: sts:AssumeRole + Condition: + StringEquals: + "sts:ExternalId": [!Ref AssumeRoleExternalId] + Description: The role Temporal Cloud uses to invoke Lambda functions for Serverless Workers. + MaxSessionDuration: 3600 + + TemporalCloudLambdaInvokePermissions: + Type: AWS::IAM::Policy + DependsOn: TemporalCloudServerlessWorker + Properties: + PolicyName: "Temporal-Cloud-Lambda-Invoke-Permissions" + PolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Action: + - lambda:InvokeFunction + - lambda:GetFunction + Resource: !Ref LambdaFunctionARNs + Roles: + - !Ref TemporalCloudServerlessWorker + +Outputs: + RoleARN: + Description: The ARN of the IAM role created for Temporal Cloud. + Value: !GetAtt TemporalCloudServerlessWorker.Arn + Export: + Name: !Sub "${AWS::StackName}-RoleARN" + + RoleName: + Description: The name of the IAM role. + Value: !Ref TemporalCloudServerlessWorker + + LambdaFunctionARNs: + Description: The Lambda function ARNs that can be invoked. + Value: !Join [", ", !Ref LambdaFunctionARNs] diff --git a/lambda-worker/otel-collector-config.template.yaml b/lambda-worker/otel-collector-config.template.yaml new file mode 100644 index 00000000..c0ccbda7 --- /dev/null +++ b/lambda-worker/otel-collector-config.template.yaml @@ -0,0 +1,33 @@ +receivers: + otlp: + protocols: + grpc: + endpoint: "localhost:4317" + http: + endpoint: "localhost:4318" + +exporters: + debug: + awsxray: + region: ${env:AWS_REGION} + awsemf: + namespace: TemporalWorkerMetrics + log_group_name: /aws/lambda/${env:AWS_LAMBDA_FUNCTION_NAME} + region: ${env:AWS_REGION} + dimension_rollup_option: NoDimensionRollup + resource_to_telemetry_conversion: + enabled: true + +service: + pipelines: + traces: + receivers: [otlp] + exporters: [awsxray, debug] + metrics: + receivers: [otlp] + exporters: [awsemf] + telemetry: + logs: + level: info + metrics: + address: localhost:8888 diff --git a/lambda-worker/starter/build.gradle b/lambda-worker/starter/build.gradle new file mode 100644 index 00000000..df615ecc --- /dev/null +++ b/lambda-worker/starter/build.gradle @@ -0,0 +1,19 @@ +dependencies { + implementation project(':lambda-worker:worker') + implementation "io.temporal:temporal-sdk:$javaSDKVersion" + implementation "io.temporal:temporal-envconfig:$javaSDKVersion" + runtimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: '1.5.6' + + dependencies { + errorproneJavac('com.google.errorprone:javac:9+181-r4173-1') + errorprone('com.google.errorprone:error_prone_core:2.28.0') + } +} + +tasks.register('execute', JavaExec) { + mainClass = 'io.temporal.samples.lambdaworker.Starter' + classpath = sourceSets.main.runtimeClasspath + if (findProperty("args")) { + args findProperty("args").tokenize() + } +} diff --git a/lambda-worker/starter/src/main/java/io/temporal/samples/lambdaworker/Starter.java b/lambda-worker/starter/src/main/java/io/temporal/samples/lambdaworker/Starter.java new file mode 100644 index 00000000..6ad4afa4 --- /dev/null +++ b/lambda-worker/starter/src/main/java/io/temporal/samples/lambdaworker/Starter.java @@ -0,0 +1,58 @@ +package io.temporal.samples.lambdaworker; + +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.envconfig.ClientConfigProfile; +import io.temporal.serviceclient.WorkflowServiceStubs; +import java.io.IOException; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +/** Local helper that starts a Workflow execution for the Lambda worker to process. */ +public class Starter { + + private static final String WORKFLOW_ID_PREFIX_ENV = "TEMPORAL_LAMBDA_WORKFLOW_ID_PREFIX"; + private static final String DEFAULT_WORKFLOW_ID_PREFIX = "serverless-workflow-id-java"; + + public static void main(String[] args) { + String name = args.length > 0 ? String.join(" ", args) : "Serverless Lambda Worker!"; + + WorkflowServiceStubs service = null; + try { + ClientConfigProfile profile = ClientConfigProfile.load(); + service = WorkflowServiceStubs.newServiceStubs(profile.toWorkflowServiceStubsOptions()); + WorkflowClient client = + WorkflowClient.newInstance(service, profile.toWorkflowClientOptions()); + + SampleWorkflow workflow = + client.newWorkflowStub( + SampleWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId(workflowIdPrefix() + "-" + UUID.randomUUID()) + .setTaskQueue(LambdaWorkerSample.taskQueue()) + .build()); + + WorkflowExecution execution = WorkflowClient.start(workflow::getGreeting, name); + System.out.printf( + "Started workflow WorkflowID=%s RunID=%s%n", + execution.getWorkflowId(), execution.getRunId()); + + String result = WorkflowStub.fromTyped(workflow).getResult(String.class); + System.out.println("Workflow result: " + result); + } catch (IOException e) { + throw new RuntimeException("Failed to load Temporal client configuration", e); + } finally { + if (service != null) { + service.shutdown(); + service.awaitTermination(10, TimeUnit.SECONDS); + } + } + } + + private static String workflowIdPrefix() { + String value = System.getenv(WORKFLOW_ID_PREFIX_ENV); + return value == null || value.isBlank() ? DEFAULT_WORKFLOW_ID_PREFIX : value; + } +} diff --git a/lambda-worker/temporal.template.toml b/lambda-worker/temporal.template.toml new file mode 100644 index 00000000..d478e3eb --- /dev/null +++ b/lambda-worker/temporal.template.toml @@ -0,0 +1,10 @@ +[profile.default] +address = "..tmprl.cloud:7233" +namespace = "." + +# Set TEMPORAL_API_KEY in the environment instead of storing it in this file. + +# For mTLS instead of API key auth, leave TEMPORAL_API_KEY unset and uncomment: +# [profile.default.tls] +# client_cert_path = "client.pem" +# client_key_path = "client.key" diff --git a/lambda-worker/worker/build.gradle b/lambda-worker/worker/build.gradle new file mode 100644 index 00000000..c6dcc092 --- /dev/null +++ b/lambda-worker/worker/build.gradle @@ -0,0 +1,58 @@ +dependencies { + implementation "io.temporal:temporal-sdk:$javaSDKVersion" + implementation "io.temporal:temporal-envconfig:$javaSDKVersion" + implementation "io.temporal:temporal-aws-lambda:$javaSDKVersion" + implementation group: 'ch.qos.logback', name: 'logback-classic', version: '1.5.6' + + testImplementation "io.temporal:temporal-testing:$javaSDKVersion" + testImplementation "junit:junit:4.13.2" + testImplementation(platform("org.junit:junit-bom:5.10.3")) + testRuntimeOnly "org.junit.vintage:junit-vintage-engine" + + dependencies { + errorproneJavac('com.google.errorprone:javac:9+181-r4173-1') + errorprone('com.google.errorprone:error_prone_core:2.28.0') + } +} + +tasks.register('shadowJar', Jar) { + archiveBaseName = 'lambda-worker' + archiveClassifier = 'all' + archiveVersion = jarVersion + duplicatesStrategy = DuplicatesStrategy.EXCLUDE + dependsOn configurations.runtimeClasspath + def mergedServicesDir = layout.buildDirectory.dir('generated/mergedServices') + + from sourceSets.main.output + from(file('../otel-collector-config.template.yaml')) { + rename 'otel-collector-config.template.yaml', 'otel-collector-config.yaml' + } + from({ + configurations.runtimeClasspath.collect { it.isDirectory() ? it : zipTree(it) } + }) { + exclude 'META-INF/services/**' + } + from(mergedServicesDir) + + doFirst { + File servicesOutput = mergedServicesDir.get().asFile + delete servicesOutput + File servicesDir = new File(servicesOutput, 'META-INF/services') + servicesDir.mkdirs() + + Map> services = [:].withDefault { new LinkedHashSet<>() } + configurations.runtimeClasspath.files.findAll { it.isFile() }.each { artifact -> + zipTree(artifact).matching { include 'META-INF/services/*' }.files.each { serviceFile -> + serviceFile.eachLine('UTF-8') { line -> + String service = line.trim() + if (!service.isEmpty() && !service.startsWith('#')) { + services[serviceFile.name].add(service) + } + } + } + } + services.each { name, providers -> + new File(servicesDir, name).text = providers.join(System.lineSeparator()) + System.lineSeparator() + } + } +} diff --git a/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/GreetingActivities.java b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/GreetingActivities.java new file mode 100644 index 00000000..620b7212 --- /dev/null +++ b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/GreetingActivities.java @@ -0,0 +1,12 @@ +package io.temporal.samples.lambdaworker; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; + +/** Activity interface used by the sample workflow. */ +@ActivityInterface +public interface GreetingActivities { + + @ActivityMethod + String createGreeting(String name); +} diff --git a/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/GreetingActivitiesImpl.java b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/GreetingActivitiesImpl.java new file mode 100644 index 00000000..07349515 --- /dev/null +++ b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/GreetingActivitiesImpl.java @@ -0,0 +1,20 @@ +package io.temporal.samples.lambdaworker; + +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityInfo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Activity implementation that returns a simple greeting. */ +public class GreetingActivitiesImpl implements GreetingActivities { + + private static final Logger logger = LoggerFactory.getLogger(GreetingActivitiesImpl.class); + + @Override + public String createGreeting(String name) { + ActivityInfo info = Activity.getExecutionContext().getInfo(); + logger.info( + "Running activity {} for workflow {}", info.getActivityType(), info.getWorkflowId()); + return "Hello, " + name + "!"; + } +} diff --git a/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/LambdaFunction.java b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/LambdaFunction.java new file mode 100644 index 00000000..6ff29cda --- /dev/null +++ b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/LambdaFunction.java @@ -0,0 +1,27 @@ +package io.temporal.samples.lambdaworker; + +import com.amazonaws.services.lambda.runtime.Context; +import com.amazonaws.services.lambda.runtime.RequestHandler; +import io.temporal.aws.lambda.LambdaWorker; +import io.temporal.aws.lambda.OtelLambdaWorkerConfigurationHelper; +import io.temporal.common.WorkerDeploymentVersion; + +/** AWS Lambda entry point for the Temporal worker. */ +public class LambdaFunction implements RequestHandler { + + private static final RequestHandler WORKER = + LambdaWorker.define( + new WorkerDeploymentVersion( + LambdaWorkerSample.deploymentName(), LambdaWorkerSample.buildId()), + builder -> { + OtelLambdaWorkerConfigurationHelper.configure(builder); + builder.setTaskQueue(LambdaWorkerSample.taskQueue()); + builder.registerWorkflowImplementationTypes(SampleWorkflowImpl.class); + builder.registerActivitiesImplementations(new GreetingActivitiesImpl()); + }); + + @Override + public Void handleRequest(Object input, Context context) { + return WORKER.handleRequest(input, context); + } +} diff --git a/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/LambdaWorkerSample.java b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/LambdaWorkerSample.java new file mode 100644 index 00000000..b24c87ed --- /dev/null +++ b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/LambdaWorkerSample.java @@ -0,0 +1,32 @@ +package io.temporal.samples.lambdaworker; + +/** Shared configuration for the Lambda Worker sample. */ +public final class LambdaWorkerSample { + + public static final String TASK_QUEUE_ENV = "TEMPORAL_TASK_QUEUE"; + public static final String DEPLOYMENT_NAME_ENV = "TEMPORAL_LAMBDA_DEPLOYMENT_NAME"; + public static final String BUILD_ID_ENV = "TEMPORAL_LAMBDA_BUILD_ID"; + + public static final String DEFAULT_TASK_QUEUE = "serverless-task-queue-java"; + public static final String DEFAULT_DEPLOYMENT_NAME = "my-app"; + public static final String DEFAULT_BUILD_ID = "build-1"; + + public static String taskQueue() { + return envOrDefault(TASK_QUEUE_ENV, DEFAULT_TASK_QUEUE); + } + + public static String deploymentName() { + return envOrDefault(DEPLOYMENT_NAME_ENV, DEFAULT_DEPLOYMENT_NAME); + } + + public static String buildId() { + return envOrDefault(BUILD_ID_ENV, DEFAULT_BUILD_ID); + } + + private static String envOrDefault(String name, String defaultValue) { + String value = System.getenv(name); + return value == null || value.isBlank() ? defaultValue : value; + } + + private LambdaWorkerSample() {} +} diff --git a/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/SampleWorkflow.java b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/SampleWorkflow.java new file mode 100644 index 00000000..25aa58f8 --- /dev/null +++ b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/SampleWorkflow.java @@ -0,0 +1,12 @@ +package io.temporal.samples.lambdaworker; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +/** Sample workflow run by the Lambda worker. */ +@WorkflowInterface +public interface SampleWorkflow { + + @WorkflowMethod + String getGreeting(String name); +} diff --git a/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/SampleWorkflowImpl.java b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/SampleWorkflowImpl.java new file mode 100644 index 00000000..0f6e7cbb --- /dev/null +++ b/lambda-worker/worker/src/main/java/io/temporal/samples/lambdaworker/SampleWorkflowImpl.java @@ -0,0 +1,28 @@ +package io.temporal.samples.lambdaworker; + +import io.temporal.activity.ActivityOptions; +import io.temporal.common.VersioningBehavior; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowVersioningBehavior; +import java.time.Duration; +import org.slf4j.Logger; + +/** Workflow implementation that executes a greeting Activity. */ +public class SampleWorkflowImpl implements SampleWorkflow { + + private static final Logger logger = Workflow.getLogger(SampleWorkflowImpl.class); + + private final GreetingActivities activities = + Workflow.newActivityStub( + GreetingActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + + @Override + @WorkflowVersioningBehavior(VersioningBehavior.PINNED) + public String getGreeting(String name) { + logger.info("SampleWorkflow started for {}", name); + String result = activities.createGreeting(name); + logger.info("SampleWorkflow completed with {}", result); + return result; + } +} diff --git a/lambda-worker/worker/src/main/resources/logback.xml b/lambda-worker/worker/src/main/resources/logback.xml new file mode 100644 index 00000000..7c5faa44 --- /dev/null +++ b/lambda-worker/worker/src/main/resources/logback.xml @@ -0,0 +1,15 @@ + + + + %d{HH:mm:ss.SSS} %-5level [%thread] %logger{36} - %msg%n + + + + + + + + + + + diff --git a/lambda-worker/worker/src/test/java/io/temporal/samples/lambdaworker/SampleWorkflowTest.java b/lambda-worker/worker/src/test/java/io/temporal/samples/lambdaworker/SampleWorkflowTest.java new file mode 100644 index 00000000..d4d2367b --- /dev/null +++ b/lambda-worker/worker/src/test/java/io/temporal/samples/lambdaworker/SampleWorkflowTest.java @@ -0,0 +1,32 @@ +package io.temporal.samples.lambdaworker; + +import static org.junit.Assert.assertEquals; + +import io.temporal.client.WorkflowOptions; +import io.temporal.testing.TestWorkflowRule; +import org.junit.Rule; +import org.junit.Test; + +/** Unit test for the sample Workflow and Activity. */ +public class SampleWorkflowTest { + + @Rule + public TestWorkflowRule testWorkflowRule = + TestWorkflowRule.newBuilder() + .setWorkflowTypes(SampleWorkflowImpl.class) + .setActivityImplementations(new GreetingActivitiesImpl()) + .build(); + + @Test + public void workflowReturnsGreeting() { + SampleWorkflow workflow = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + SampleWorkflow.class, + WorkflowOptions.newBuilder().setTaskQueue(testWorkflowRule.getTaskQueue()).build()); + + assertEquals( + "Hello, Serverless Lambda Worker!!", workflow.getGreeting("Serverless Lambda Worker!")); + } +} diff --git a/settings.gradle b/settings.gradle index b99ad281..b19f2fd7 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,4 +1,5 @@ rootProject.name = 'temporal-java-samples' + include 'core' include 'springai:basic' include 'springai:mcp' @@ -6,4 +7,5 @@ include 'springai:multimodel' include 'springai:rag' include 'springboot' include 'springboot-basic' - +include 'lambda-worker:starter' +include 'lambda-worker:worker'