Merge pull request #536 from stac-utils/pv/fix-sns-publish

fix async sns post-ingest publishing

Author: Phil Varner

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [Unreleased] - TBD
+
+### Fixed
+
+- Post-ingest SNS topic was not being published to when deployed as a Lambda.
+
 ## [2.0.0] - 2023-06-26
 
 ### Removed

README.md

@@ -8,6 +8,7 @@
   - [Migration](#migration)
     - [0.x or 1.x -\> 2.x](#0x-or-1x---2x)
       - [Fine-grained Access Control](#fine-grained-access-control)
+      - [Enabling Post-ingest SNS publishing](#enabling-post-ingest-sns-publishing)
     - [0.4.x -\> 0.5.x](#04x---05x)
       - [Elasticsearch to OpenSearch Migration](#elasticsearch-to-opensearch-migration)
       - [Preferred Elasticsearch to OpenSearch Migration Process](#preferred-elasticsearch-to-opensearch-migration-process)
@@ -135,6 +136,49 @@ As of 2.0.0, only OpenSearch is supported and only using fine-grained access con
 It is recommended to follow the migration path to upgrade to fine-grained access control
 first and then upgrade to stac-server 2.x.
 
+#### Enabling Post-ingest SNS publishing
+
+stac-server now has the ability to publish all ingested entities (Items and Collections)
+to an SNS topic. Follow these steps to add this to an existing deployment. These
+configurations are also in the serverless.example.yml file, so reference that if it is
+unclear exactly where to add this in your config.
+
+The following changes should be added to the serverless.yml file.
+
+Explicitly set the provider/environment setting for STAC_API_URL so the ingested entities
+published to the topic will have their link hrefs set correctly. If this is not set,
+the entities will still be published, with with incorrect link hrefs.
+
+```text
+STAC_API_URL: "https://some-stac-server.com"
+```
+
+Add the SNS topic resource:
+
+```text
+postIngestTopic:
+  Type: AWS::SNS::Topic
+  Properties:
+    TopicName: ${self:service}-${self:provider.stage}-post-ingest
+```
+
+For the `ingest` Lambda resource definition, configure the ARN to publish to by adding:
+
+```text
+environment:
+  POST_INGEST_TOPIC_ARN: !Ref postIngestTopic
+```
+
+Add IAM permissions with the statement:
+
+```text
+- Effect: Allow
+  Action:
+    - sns:Publish
+  Resource:
+    Fn::GetAtt: [postIngestTopic, TopicArn]
+```
+
 ### 0.4.x -> 0.5.x
 
 #### Elasticsearch to OpenSearch Migration
@@ -521,11 +565,9 @@ aws lambda invoke \
   /dev/stdout
 ```
 
-Stac-server is now ready to ingest data!
-
 #### OpenSearch fine-grained access control
 
-As of version 2.0.0, stac-server on"ly supports fine-grained access control to
+As of version 2.0.0, stac-server only supports fine-grained access control to
 OpenSearch, and no longer supports "AWS Connection" mode.
 
 **Warning**: Unfortunately, fine-grained access control cannot be enabled on an
@@ -634,8 +676,8 @@ so that stac-server can access them.
 The preferred mechanism for populating the OpenSearch credentials to stac-server is to
 create a secret in AWS Secret Manager that contains the username and password. The
 recommended name for this Secret corresponds
-to the stac-server deployment as `{stage}/{service}/opensearch`, e.g.,
-`dev/my-stac-server/opensearch`.
+to the stac-server deployment as `${service}-${stage}-opensearch-user-creds`, e.g.,
+`my-stac-server-dev-opensearch-user-creds`.
 
 The Secret type should be "Other type of secret" and
 have two keys, `username` and `password`, with the appropriate
@@ -645,14 +687,14 @@ Add the `OPENSEARCH_CREDENTIALS_SECRET_ID` variable to the serverless.yml sectio
 `environment`:
 
 ```yaml
-OPENSEARCH_CREDENTIALS_SECRET_ID: ${self:provider.stage}/${self:service}/opensearch
+OPENSEARCH_CREDENTIALS_SECRET_ID: ${self:service}-${self:provider.stage}-opensearch-user-creds
 ```
 
 Add to the IAM Role Statements:
 
 ```yaml
-- Effect: "Allow"
-  Resource: "arn:aws:secretsmanager:${aws:region}:${aws:accountId}:secret:${self:provider.stage}/${self:service}/opensearch-*"
+- Effect: Allow
+  Resource: arn:aws:secretsmanager:${aws:region}:${aws:accountId}:secret:${self:provider.environment.OPENSEARCH_CREDENTIALS_SECRET_ID}-*
   Action: "secretsmanager:GetSecretValue"
 ```
 
@@ -674,6 +716,8 @@ OPENSEARCH_PASSWORD: xxxxxxxxxxx
 Setting these as environment variables can also be useful when running stac-server
 locally.
 
+Stac-server is now ready to ingest data!
+
 ### Proxying Stac-server through CloudFront
 
 The API Gateway URL associated with the deployed stac-server instance may not be the URL that you ultimately wish to expose to your API users. AWS CloudFront can be used to proxy to a more human readable URL. In order to accomplish this:

serverless.example.yml

@@ -29,20 +29,25 @@ provider:
   iam:
     role:
       statements:
-        - Effect: "Allow"
+        - Effect: Allow
           Resource: "arn:aws:es:${aws:region}:${aws:accountId}:domain/*"
           Action: "es:*"
-        - Effect: "Allow"
+        - Effect: Allow
           Action:
             - sqs:GetQueueUrl
             - sqs:SendMessage
             - sqs:ReceiveMessage
             - sqs:DeleteMessage
           Resource:
             Fn::GetAtt: [ingestQueue, Arn]
+        - Effect: Allow
+          Action:
+            - sns:Publish
+          Resource:
+            Fn::GetAtt: [postIngestTopic, Arn]
         - Effect: Allow
           Action: s3:GetObject
-          Resource: 'arn:aws:s3:::usgs-landsat/*'
+          Resource: "arn:aws:s3:::usgs-landsat/*"
         - Effect: Allow
           Resource: arn:aws:secretsmanager:${aws:region}:${aws:accountId}:secret:${self:provider.environment.OPENSEARCH_CREDENTIALS_SECRET_ID}-*
           Action: secretsmanager:GetSecretValue
@@ -107,14 +112,14 @@ resources:
   Description: A STAC API running on stac-server
   Resources:
     ingestTopic:
-      Type: "AWS::SNS::Topic"
+      Type: AWS::SNS::Topic
       Properties:
         TopicName: ${self:service}-${self:provider.stage}-ingest
     postIngestTopic:
-    # After a collection or item is ingested, the status of the ingest (success
-    # or failure) along with details of the collection or item are sent to this
-    # SNS topic. To take future action on items after they are ingested
-    # suscribe an endpoint to this topic
+      # After a collection or item is ingested, the status of the ingest (success
+      # or failure) along with details of the collection or item are sent to this
+      # SNS topic. To take future action on items after they are ingested
+      # suscribe an endpoint to this topic
       Type: AWS::SNS::Topic
       Properties:
         TopicName: ${self:service}-${self:provider.stage}-post-ingest
@@ -156,7 +161,7 @@ resources:
         TopicArn: !Ref ingestTopic
     OpenSearchInstance:
       Type: AWS::OpenSearchService::Domain
-      DeletionPolicy : Retain
+      DeletionPolicy: Retain
       UpdateReplacePolicy: Retain
       UpdatePolicy:
         EnableVersionUpgrade: true
@@ -187,13 +192,13 @@ resources:
           MasterUserOptions:
             MasterUserName: admin
             MasterUserPassword: ${env:OPENSEARCH_MASTER_USER_PASSWORD}
-          AccessPolicies:
-            Version:                        "2012-10-17"
-            Statement:
-              - Effect:                     "Allow"
-                Principal:                  { "AWS": "*" }
-                Action:                     "es:ESHttp*"
-                Resource:                   "arn:aws:es:${aws:region}:${aws:accountId}:domain/${self:service}-${self:provider.stage}/*"
+        AccessPolicies:
+          Version: "2012-10-17"
+          Statement:
+            - Effect: "Allow"
+              Principal: { "AWS": "*" }
+              Action: "es:ESHttp*"
+              Resource: "arn:aws:es:${aws:region}:${aws:accountId}:domain/${self:service}-${self:provider.stage}/*"
   Outputs:
     OpenSearchEndpoint:
       Value:

src/lambdas/ingest/index.js

@@ -53,6 +53,7 @@ export const handler = async (event, _context) => {
 
   if (event.create_indices) {
     await createIndex('collections')
+    return
   }
 
   const stacItems = isSqsEvent(event)
@@ -67,9 +68,9 @@ export const handler = async (event, _context) => {
 
     if (postIngestTopicArn) {
       logger.debug('Publishing to post-ingest topic: %s', postIngestTopicArn)
-      publishResultsToSns(results, postIngestTopicArn)
+      await publishResultsToSns(results, postIngestTopicArn)
     } else {
-      logger.debug('Skkipping post-ingest notification since no topic is configured')
+      logger.debug('Skipping post-ingest notification since no topic is configured')
     }
   } catch (error) {
     logger.error(error)

src/lib/ingest.js

@@ -28,13 +28,13 @@ export async function convertIngestObjectToDbObject(
     index = data.collection
   } else {
     throw new InvalidIngestError(
-      `Expeccted data.type to be "Collection" or "Feature" not ${data.type}`
+      `Expected data.type to be "Collection" or "Feature" not ${data.type}`
     )
   }
 
   // remove any hierarchy links in a non-mutating way
   if (!data.links) {
-    throw new InvalidIngestError('Expected a "links" proporty on the stac object')
+    throw new InvalidIngestError('Expected a "links" property on the stac object')
   }
   const links = data.links.filter(
     (/** @type {{ rel: string; }} */ link) => !hierarchyLinks.includes(link.rel)
@@ -187,11 +187,11 @@ function updateLinksWithinRecord(record) {
   return record
 }
 
-export function publishResultsToSns(results, topicArn) {
-  results.forEach(async (result) => {
+export async function publishResultsToSns(results, topicArn) {
+  await Promise.allSettled(results.map(async (result) => {
     if (result.record && !result.error) {
       updateLinksWithinRecord(result.record)
     }
     await publishRecordToSns(topicArn, result.record, result.error)
-  })
+  }))
 }