diff --git a/s3-opensearch-serverless-osis-sam/README.md b/s3-opensearch-serverless-osis-sam/README.md new file mode 100644 index 000000000..00ac91fc1 --- /dev/null +++ b/s3-opensearch-serverless-osis-sam/README.md @@ -0,0 +1,90 @@ +# Amazon S3 to Amazon OpenSearch Serverless with OpenSearch Ingestion + +This sample project demonstrates how to ingest JSON objects from an Amazon S3 bucket into an Amazon OpenSearch Serverless collection using an Amazon OpenSearch Ingestion (OSIS) pipeline with an S3 source. + +Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/s3-opensearch-serverless-osis-sam + +![Architecture diagram](images/architecture.png) + +Important: this application uses various AWS services and there are costs associated with these services after the Free Tier usage - please see the [AWS Pricing page](https://aws.amazon.com/pricing/) for details. You are responsible for any AWS costs incurred. No warranty is implied in this example. + +## Requirements + +* [Create an AWS account](https://portal.aws.amazon.com/gp/aws/developer/registration/index.html) if you do not already have one and log in. The IAM user that you use must have sufficient permissions to make necessary AWS service calls and manage AWS resources. +* [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html) installed and configured +* [Git Installed](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git) +* [AWS Serverless Application Model](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-sam-cli-install.html) (AWS SAM) installed + +## Deployment Instructions + +1. Create a new directory, navigate to that directory in a terminal and clone the GitHub repository: + ``` + git clone https://github.com/aws-samples/serverless-patterns + ``` +1. Change directory to the pattern directory: + ``` + cd s3-opensearch-serverless-osis-sam + ``` +1. From the command line, use AWS SAM to deploy the AWS resources for the pattern as specified in the template.yaml file: + ``` + sam deploy --capabilities CAPABILITY_NAMED_IAM --guided + ``` + +1. Identify the ARN of the role (or user) being used to access the console for this deployment. If it does not have the `AdministratorAccess` AWS managed policy (or a similarly permissive equivalent), attach the `AmazonOpenSearchServiceFullAccess` AWS managed policy. Access for allowed principals can be edited later via the `DataAccessPolicyEditURL` from the stack outputs. + +1. During the prompts: + * Enter a stack name + * Enter the desired AWS Region + * Enter the ARN identified in the previous step in the `AccessARN` parameter + * Enter the `DeploymentName` parameter (up to 32 characters without spaces; "a-z", "0-9", or "-") + * Allow SAM CLI to create IAM roles with the required permissions. + + Once you have run `sam deploy --guided` mode once and saved arguments to a configuration file (samconfig.toml), you can use `sam deploy` in future to use these defaults. + +1. Note the outputs from the SAM deployment process. These contain the resource names and URLs which are used for testing. + +## How it works + +* A source S3 bucket is configured to publish `s3:ObjectCreated:*` event notifications to an Amazon SQS queue. +* An OpenSearch Ingestion (OSIS) pipeline uses the [Amazon S3 source](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/configure-client-s3.html). It polls the SQS queue, and when a message indicates a new object was created, it reads the object from the source bucket. +* Each object is expected to be newline-delimited JSON (JSON Lines). The pipeline reads it line by line (`newline` codec) and the `parse_json` processor converts each line into a structured document. +* The documents are indexed into the OpenSearch Serverless collection (the `serverless: true` flag is set on the sink), into an index named `s3-logs`. +* Any documents that fail to be indexed are written to a separate dead-letter S3 bucket. A separate DLQ bucket is used deliberately: writing DLQ objects back into the source bucket would re-trigger the S3 to SQS notification and create an ingestion loop. + +The pipeline assumes an IAM role that grants it read access to the source bucket, consume access to the SQS queue, write access to the DLQ bucket, and data access to the OpenSearch Serverless collection. + +## Testing + +A sample newline-delimited JSON (JSON Lines) file, `sample.json`, is included in this pattern directory. Each line is a single JSON document: + +```json +{"timestamp": "2024-01-01T10:00:00Z", "level": "INFO", "service": "checkout", "message": "order placed", "order_id": "1001"} +{"timestamp": "2024-01-01T10:00:05Z", "level": "ERROR", "service": "checkout", "message": "payment failed", "order_id": "1002"} +{"timestamp": "2024-01-01T10:00:09Z", "level": "INFO", "service": "shipping", "message": "label created", "order_id": "1001"} +``` + +1. Upload the included `sample.json` to the source bucket (`DataBucketName` from the stack outputs). Replace the bucket name and region with your values: + ```bash + aws s3 cp sample.json s3:/// --region + ``` + +1. Wait a short time for the pipeline to poll the SQS queue and ingest the object (typically under a minute). + +1. From the OpenSearch dashboard Dev Tools console (`DevToolsURL` from the stack outputs), search the index. You should see the three documents from the uploaded file, each parsed into structured fields with the S3 source metadata (`bucket` and `key`) added by the pipeline: + ``` + GET s3-logs/_search + ``` + +## Cleanup + +1. From the [S3 console](https://console.aws.amazon.com/s3/home), empty both buckets created by this project (`DataBucketName` and `DlqBucketName` from the stack outputs). + +1. Delete the stack: + ```bash + sam delete + ``` + +---- +Copyright 2025 Amazon.com, Inc. or its affiliates. All Rights Reserved. + +SPDX-License-Identifier: MIT-0 diff --git a/s3-opensearch-serverless-osis-sam/example-pattern.json b/s3-opensearch-serverless-osis-sam/example-pattern.json new file mode 100644 index 000000000..1a5d62479 --- /dev/null +++ b/s3-opensearch-serverless-osis-sam/example-pattern.json @@ -0,0 +1,64 @@ +{ + "title": "Amazon S3 to Amazon OpenSearch Serverless with OpenSearch Ingestion", + "description": "Ingest JSON objects from Amazon S3 into Amazon OpenSearch Serverless using an Amazon OpenSearch Ingestion (OSIS) pipeline.", + "language": "YAML", + "level": "300", + "framework": "AWS SAM", + "introBox": { + "headline": "How it works", + "text": [ + "This pattern shows how to build a fully serverless ingestion path from Amazon S3 into an Amazon OpenSearch Serverless collection using an Amazon OpenSearch Ingestion (OSIS) pipeline. OpenSearch Ingestion is a fully managed, serverless data collector built on Data Prepper that ingests, filters, transforms, and routes data to an OpenSearch destination without you having to run or scale any ingestion infrastructure. It is a common building block for log analytics, search indexing, and observability workloads where files land in Amazon S3 and need to be searchable in near real time.", + "The flow works as follows. A client uploads a newline-delimited JSON (JSON Lines) object to the source S3 bucket. The bucket is configured to publish an s3:ObjectCreated event notification to an Amazon SQS queue. The OSIS pipeline uses the Amazon S3 source, which continuously polls the SQS queue; when a message indicates that a new object was created, the pipeline reads that object from the bucket. The pipeline parses the object one line at a time using the newline codec, and a parse_json processor converts each line into a structured document. The documents are then indexed into the OpenSearch Serverless collection through the pipeline's OpenSearch sink, which sets the serverless flag and authenticates to the collection with AWS SigV4.", + "Security and reliability are built in. The pipeline assumes a dedicated IAM role that is scoped to read the source bucket, consume from the SQS queue, write to the dead-letter bucket, and access the collection. The OpenSearch Serverless collection is protected by encryption, network, and data-access policies, and only the pipeline role and the principal you supply at deploy time are granted data access. Documents that fail to be indexed are written to a separate dead-letter S3 (DLQ) bucket. Using a separate DLQ bucket is a deliberate design choice: writing failed records back into the source bucket would re-trigger the S3 to SQS notification and create an ingestion loop. The pipeline also publishes audit logs to Amazon CloudWatch Logs and scales automatically between the configured minimum and maximum OpenSearch Compute Units (OCUs).", + "This pattern deploys two Amazon S3 buckets (source and dead-letter), two Amazon SQS queues (an ingest queue and its dead-letter queue), an Amazon OpenSearch Serverless collection with its encryption, network, and data-access policies, an Amazon OpenSearch Ingestion pipeline, a CloudWatch log group, and the IAM role that the pipeline assumes. It is deployed with AWS SAM and requires no custom application code. Important: this example uses AWS services that may incur costs beyond the Free Tier, and the network policy allows public access for demo simplicity; for production, restrict access using a VPC endpoint." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/s3-opensearch-serverless-osis-sam", + "templateURL": "serverless-patterns/s3-opensearch-serverless-osis-sam", + "projectFolder": "s3-opensearch-serverless-osis-sam", + "templateFile": "template.yaml" + } + }, + "resources": { + "bullets": [ + { + "text": "Using an OpenSearch Ingestion pipeline with Amazon S3", + "link": "https://docs.aws.amazon.com/opensearch-service/latest/developerguide/configure-client-s3.html" + }, + { + "text": "Ingesting data into Amazon OpenSearch Serverless collections", + "link": "https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-clients.html" + }, + { + "text": "Amazon OpenSearch Ingestion", + "link": "https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ingestion.html" + } + ] + }, + "deploy": { + "text": [ + "sam deploy --guided" + ] + }, + "testing": { + "text": [ + "See the GitHub repo for detailed testing instructions." + ] + }, + "cleanup": { + "text": [ + "Empty both S3 buckets, then delete the stack: sam delete." + ] + }, + "authors": [ + { + "name": "Sudhansu", + "image": "https://drive.google.com/file/d/1uTQ3MbFvziRfBPWHsVsD5vbVyGuFO665/view?usp=sharing", + "bio": "Serverless builder contributing to Serverless Land patterns.", + "linkedin": "sudhansu-mohapatra-83a51811a", + "twitter": "" + } + ] +} diff --git a/s3-opensearch-serverless-osis-sam/images/architecture.png b/s3-opensearch-serverless-osis-sam/images/architecture.png new file mode 100644 index 000000000..155867229 Binary files /dev/null and b/s3-opensearch-serverless-osis-sam/images/architecture.png differ diff --git a/s3-opensearch-serverless-osis-sam/sample.json b/s3-opensearch-serverless-osis-sam/sample.json new file mode 100644 index 000000000..7610e3783 --- /dev/null +++ b/s3-opensearch-serverless-osis-sam/sample.json @@ -0,0 +1,3 @@ +{"timestamp": "2024-01-01T10:00:00Z", "level": "INFO", "service": "checkout", "message": "order placed", "order_id": "1001"} +{"timestamp": "2024-01-01T10:00:05Z", "level": "ERROR", "service": "checkout", "message": "payment failed", "order_id": "1002"} +{"timestamp": "2024-01-01T10:00:09Z", "level": "INFO", "service": "shipping", "message": "label created", "order_id": "1001"} diff --git a/s3-opensearch-serverless-osis-sam/template.yaml b/s3-opensearch-serverless-osis-sam/template.yaml new file mode 100644 index 000000000..5d7746e35 --- /dev/null +++ b/s3-opensearch-serverless-osis-sam/template.yaml @@ -0,0 +1,320 @@ +--- +AWSTemplateFormatVersion: "2010-09-09" +Transform: AWS::Serverless-2016-10-31 +Description: >- + Serverless patterns - Amazon S3 to Amazon OpenSearch Serverless using an + Amazon OpenSearch Ingestion (OSIS) pipeline with an S3 source. + (tag:s3-opensearch-serverless-osis-sam) + +Parameters: + AccessARN: + Description: >- + ARN of the IAM role or user that will access the OpenSearch dashboard / + collection (for example, the principal you use in the AWS console). + Type: String + DeploymentName: + Description: Unique lowercase value used to name the resources in this stack. + Default: demo-s3-oss-osis + Type: String + AllowedPattern: ^[a-z][a-z0-9-]{2,31}$ + +Resources: + + # --------------------------------------------------------------------------- + # Source bucket. Users upload newline-delimited JSON (JSON Lines) log files + # here. Object-created events are published to the ingestion SQS queue. + # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-s3-bucket.html + # --------------------------------------------------------------------------- + DataBucket: + Type: AWS::S3::Bucket + DependsOn: + - IngestQueuePolicy + Properties: + # Fixed name (built from DeploymentName + account + region) so the queue + # policy can reference the bucket ARN without creating a circular + # dependency between the bucket notification and the queue policy. + BucketName: !Sub "${DeploymentName}-data-${AWS::AccountId}-${AWS::Region}" + BucketEncryption: + ServerSideEncryptionConfiguration: + - BucketKeyEnabled: true + ServerSideEncryptionByDefault: + SSEAlgorithm: AES256 + PublicAccessBlockConfiguration: + BlockPublicAcls: true + BlockPublicPolicy: true + IgnorePublicAcls: true + RestrictPublicBuckets: true + NotificationConfiguration: + QueueConfigurations: + - Event: "s3:ObjectCreated:*" + Queue: !GetAtt IngestQueue.Arn + + # --------------------------------------------------------------------------- + # Separate bucket for the pipeline dead-letter queue (DLQ). Kept separate from + # DataBucket on purpose: writing DLQ objects into the source bucket would + # re-trigger the S3 -> SQS notification and create an ingestion loop. + # --------------------------------------------------------------------------- + DlqBucket: + Type: AWS::S3::Bucket + Properties: + BucketName: !Sub "${DeploymentName}-dlq-${AWS::AccountId}-${AWS::Region}" + BucketEncryption: + ServerSideEncryptionConfiguration: + - BucketKeyEnabled: true + ServerSideEncryptionByDefault: + SSEAlgorithm: AES256 + PublicAccessBlockConfiguration: + BlockPublicAcls: true + BlockPublicPolicy: true + IgnorePublicAcls: true + RestrictPublicBuckets: true + + # --------------------------------------------------------------------------- + # SQS queue that receives S3 object-created notifications. The OSIS S3 source + # polls this queue to discover new objects to ingest. + # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-sqs-queue.html + # --------------------------------------------------------------------------- + IngestQueue: + Type: AWS::SQS::Queue + Properties: + QueueName: !Sub "${DeploymentName}-ingest" + # Visibility timeout should be >= the time the pipeline needs to process a + # batch of objects before the message becomes visible again. + VisibilityTimeout: 90 + MessageRetentionPeriod: 345600 + RedrivePolicy: + deadLetterTargetArn: !GetAtt IngestDlq.Arn + maxReceiveCount: 5 + + # Dead-letter queue for SQS notification messages that repeatedly fail. + IngestDlq: + Type: AWS::SQS::Queue + Properties: + QueueName: !Sub "${DeploymentName}-ingest-dlq" + MessageRetentionPeriod: 1209600 + + # Allow Amazon S3 to send object-created notifications to the ingest queue. + # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-sqs-queuepolicy.html + IngestQueuePolicy: + Type: AWS::SQS::QueuePolicy + Properties: + Queues: + - !Ref IngestQueue + PolicyDocument: + Version: "2012-10-17" + Statement: + - Sid: AllowS3SendMessage + Effect: Allow + Principal: + Service: s3.amazonaws.com + Action: "sqs:SendMessage" + Resource: !GetAtt IngestQueue.Arn + Condition: + ArnLike: + "aws:SourceArn": !Sub "arn:${AWS::Partition}:s3:::${DeploymentName}-data-${AWS::AccountId}-${AWS::Region}" + StringEquals: + "aws:SourceAccount": !Ref "AWS::AccountId" + + # --------------------------------------------------------------------------- + # IAM role assumed by the OpenSearch Ingestion pipeline. It needs to read the + # source bucket, consume from the SQS queue, write to the DLQ bucket, and + # send documents to the OpenSearch Serverless collection. + # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-iam-role.html + # --------------------------------------------------------------------------- + PipelineRole: + Type: AWS::IAM::Role + Properties: + RoleName: !Sub "${DeploymentName}-pipeline" + AssumeRolePolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Principal: + Service: + - osis-pipelines.amazonaws.com + Action: + - "sts:AssumeRole" + Policies: + - PolicyName: OpenSearchServerless + PolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Action: + - "aoss:BatchGetCollection" + - "aoss:APIAccessAll" + Resource: !GetAtt OpenSearchServerless.Arn + - Effect: Allow + Action: + - "aoss:CreateSecurityPolicy" + - "aoss:GetSecurityPolicy" + - "aoss:UpdateSecurityPolicy" + Resource: "*" + Condition: + StringEquals: + "aoss:collection": !Ref DeploymentName + - PolicyName: SourceBucketRead + PolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Action: + - "s3:GetObject" + Resource: !Sub "arn:${AWS::Partition}:s3:::${DeploymentName}-data-${AWS::AccountId}-${AWS::Region}/*" + - Effect: Allow + Action: + - "s3:ListBucket" + Resource: !Sub "arn:${AWS::Partition}:s3:::${DeploymentName}-data-${AWS::AccountId}-${AWS::Region}" + - PolicyName: DlqBucketWrite + PolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Action: + - "s3:PutObject" + - "s3:AbortMultipartUpload" + Resource: !Sub "${DlqBucket.Arn}/*" + - PolicyName: SqsConsume + PolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Action: + - "sqs:ReceiveMessage" + - "sqs:DeleteMessage" + - "sqs:GetQueueAttributes" + - "sqs:ChangeMessageVisibility" + Resource: !GetAtt IngestQueue.Arn + + # --------------------------------------------------------------------------- + # OpenSearch Serverless collection and its required security policies. + # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-opensearchserverless-collection.html + # --------------------------------------------------------------------------- + OpenSearchServerless: + Type: AWS::OpenSearchServerless::Collection + DependsOn: + - EncryptionPolicy + - NetworkPolicy + - AccessPolicy + Properties: + Name: !Ref DeploymentName + Type: SEARCH + + EncryptionPolicy: + Type: AWS::OpenSearchServerless::SecurityPolicy + Properties: + Name: !Ref DeploymentName + Type: encryption + Description: Encryption policy for the collection. + Policy: !Sub >- + {"Rules":[{"ResourceType":"collection","Resource":["collection/${DeploymentName}"]}],"AWSOwnedKey":true} + + NetworkPolicy: + Type: AWS::OpenSearchServerless::SecurityPolicy + Properties: + Name: !Ref DeploymentName + Type: network + Description: Network policy for the collection. + # AllowFromPublic is used here for a simple demo. For production, restrict + # access using a VPC endpoint (SourceVPCEs) instead. + Policy: !Sub >- + [{"Rules":[{"ResourceType":"collection","Resource":["collection/${DeploymentName}"]},{"ResourceType":"dashboard","Resource":["collection/${DeploymentName}"]}],"AllowFromPublic":true}] + + AccessPolicy: + Type: AWS::OpenSearchServerless::AccessPolicy + Properties: + Name: !Ref DeploymentName + Type: data + Description: Data access policy for the pipeline role and the console principal. + Policy: !Sub >- + [{"Description":"Access for pipeline and admin","Rules":[{"ResourceType":"index","Resource":["index/${DeploymentName}/*"],"Permission":["aoss:*"]},{"ResourceType":"collection","Resource":["collection/${DeploymentName}"],"Permission":["aoss:*"]}],"Principal":["${AccessARN}","arn:${AWS::Partition}:iam::${AWS::AccountId}:role/${DeploymentName}-pipeline"]}] + + # --------------------------------------------------------------------------- + # CloudWatch log group for pipeline audit logs. + # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-logs-loggroup.html + # --------------------------------------------------------------------------- + CloudWatchLogGroup: + Type: AWS::Logs::LogGroup + Properties: + LogGroupName: !Sub "/aws/vendedlogs/OpenSearchIngestion/${DeploymentName}/audit-logs" + RetentionInDays: 3 + + # --------------------------------------------------------------------------- + # OpenSearch Ingestion (OSIS) pipeline. Reads objects announced on the SQS + # queue from the source bucket, parses each JSON line, and indexes the + # documents into the OpenSearch Serverless collection. + # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-osis-pipeline.html + # --------------------------------------------------------------------------- + OpenSearchPipeline: + Type: AWS::OSIS::Pipeline + Properties: + PipelineName: !Ref DeploymentName + LogPublishingOptions: + CloudWatchLogDestination: + LogGroup: !Ref CloudWatchLogGroup + IsLoggingEnabled: true + MaxUnits: 4 + MinUnits: 1 + PipelineConfigurationBody: !Sub + | + version: "2" + s3-log-pipeline: + source: + s3: + # Acknowledge SQS messages only after data reaches the sink. + acknowledgments: true + notification_type: "sqs" + compression: "none" + # Treat each line of the object as one record; the parse_json + # processor below converts each line into a structured document. + codec: + newline: + sqs: + queue_url: "${IngestQueue}" + maximum_messages: 10 + visibility_timeout: "90s" + aws: + region: "${AWS::Region}" + sts_role_arn: "${PipelineRole.Arn}" + processor: + - parse_json: + source: "message" + sink: + - opensearch: + hosts: + [ + "${OpenSearchServerless.CollectionEndpoint}" + ] + index: "s3-logs" + aws: + sts_role_arn: "${PipelineRole.Arn}" + region: "${AWS::Region}" + # Required when the sink is an OpenSearch Serverless collection. + serverless: true + # Capture failed documents in the separate DLQ bucket. + dlq: + s3: + bucket: "${DlqBucket}" + key_path_prefix: "s3-log-pipeline/dlq" + region: "${AWS::Region}" + sts_role_arn: "${PipelineRole.Arn}" + +Outputs: + DataBucketName: + Description: Upload newline-delimited JSON (.json / .log) files here to ingest them. + Value: !Ref DataBucket + DlqBucketName: + Description: Failed documents are written here. Empty this bucket before deleting the stack. + Value: !Ref DlqBucket + IngestQueueUrl: + Description: SQS queue that receives S3 object-created notifications. + Value: !Ref IngestQueue + CollectionEndpoint: + Description: OpenSearch Serverless collection endpoint. + Value: !GetAtt OpenSearchServerless.CollectionEndpoint + DevToolsURL: + Description: Direct URL to Dev Tools within the OpenSearch dashboard to query the index. + Value: !Sub "${OpenSearchServerless.DashboardEndpoint}/app/dev_tools#/console" + DataAccessPolicyEditURL: + Description: Direct URL to modify the data access policy if you need to add principals. + Value: !Sub "https://${AWS::Region}.console.aws.amazon.com/aos/home?region=${AWS::Region}#opensearch/security/access-policies/${DeploymentName}/editAccessPolicy"