feat(ingestor-api): add StacIngestor construct

Author: alukach

lib/index.ts

@@ -1,4 +1,5 @@
 export * from "./bastion-host";
 export * from "./bootstrapper";
 export * from "./database";
+export * from "./ingestor-api";
 export * from "./stac-api";

lib/ingestor-api/index.ts

@@ -0,0 +1,256 @@
+import {
+  aws_apigateway as apigateway,
+  aws_dynamodb as dynamodb,
+  aws_ec2 as ec2,
+  aws_iam as iam,
+  aws_lambda as lambda,
+  aws_lambda_event_sources as events,
+  aws_secretsmanager as secretsmanager,
+  aws_ssm as ssm,
+  Duration,
+  RemovalPolicy,
+  Stack,
+} from "aws-cdk-lib";
+import { PythonFunction } from "@aws-cdk/aws-lambda-python-alpha";
+import { Construct } from "constructs";
+
+export class StacIngestor extends Construct {
+  table: dynamodb.Table;
+
+  constructor(scope: Construct, id: string, props: StacIngestorProps) {
+    super(scope, id);
+
+    this.table = this.buildTable();
+
+    const env: Record<string, string> = {
+      DYNAMODB_TABLE: this.table.tableName,
+      ROOT_PATH: `/${props.stage}`,
+      NO_PYDANTIC_SSM_SETTINGS: "1",
+      STAC_URL: props.stacUrl,
+      DATA_ACCESS_ROLE: props.dataAccessRole.roleArn,
+      ...props.apiEnv,
+    };
+
+    const handler = this.buildApiLambda({
+      table: this.table,
+      env,
+      dataAccessRole: props.dataAccessRole,
+      stage: props.stage,
+    });
+
+    this.buildApiEndpoint({
+      handler,
+      stage: props.stage,
+      endpointConfiguration: props.apiEndpointConfiguration,
+      policy: props.apiPolicy,
+    });
+
+    this.buildIngestor({
+      table: this.table,
+      env: env,
+      dbSecret: props.stacDbSecret,
+      dbVpc: props.vpc,
+      dbSecurityGroup: props.stacDbSecurityGroup,
+      subnetSelection: props.subnetSelection,
+    });
+
+    this.registerSsmParameter({
+      name: "dynamodb_table",
+      value: this.table.tableName,
+      description: "Name of table used to store ingestions",
+    });
+  }
+
+  private buildTable(): dynamodb.Table {
+    const table = new dynamodb.Table(this, "ingestions-table", {
+      partitionKey: { name: "created_by", type: dynamodb.AttributeType.STRING },
+      sortKey: { name: "id", type: dynamodb.AttributeType.STRING },
+      billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
+      removalPolicy: RemovalPolicy.DESTROY,
+      stream: dynamodb.StreamViewType.NEW_IMAGE,
+    });
+
+    table.addGlobalSecondaryIndex({
+      indexName: "status",
+      partitionKey: { name: "status", type: dynamodb.AttributeType.STRING },
+      sortKey: { name: "created_at", type: dynamodb.AttributeType.STRING },
+    });
+
+    return table;
+  }
+
+  private buildApiLambda(props: {
+    table: dynamodb.ITable;
+    env: Record<string, string>;
+    dataAccessRole: iam.IRole;
+    stage: string;
+  }): PythonFunction {
+    const handler_role = new iam.Role(this, "execution-role", {
+      description:
+        "Role used by STAC Ingestor. Manually defined so that we can choose a name that is supported by the data access roles trust policy",
+      roleName: `stac-ingestion-api-${props.stage}`,
+      assumedBy: new iam.ServicePrincipal("lambda.amazonaws.com"),
+      managedPolicies: [
+        iam.ManagedPolicy.fromAwsManagedPolicyName(
+          "service-role/AWSLambdaBasicExecutionRole"
+        ),
+      ],
+    });
+
+    const handler = new PythonFunction(this, "api-handler", {
+      entry: `${__dirname}/runtime`,
+      index: "src/handler.py",
+      runtime: lambda.Runtime.PYTHON_3_9,
+      environment: props.env,
+      timeout: Duration.seconds(30),
+      role: handler_role,
+      memorySize: 2048,
+    });
+
+    props.table.grantReadWriteData(handler);
+    props.dataAccessRole.grant(handler.grantPrincipal, "sts:AssumeRole");
+
+    return handler;
+  }
+
+  private buildIngestor(props: {
+    table: dynamodb.ITable;
+    env: Record<string, string>;
+    dbSecret: secretsmanager.ISecret;
+    dbVpc: ec2.IVpc;
+    dbSecurityGroup: ec2.ISecurityGroup;
+    subnetSelection: ec2.SubnetSelection;
+  }): PythonFunction {
+    const handler = new PythonFunction(this, "stac-ingestor", {
+      entry: `${__dirname}/runtime`,
+      index: "src/ingestor.py",
+      runtime: lambda.Runtime.PYTHON_3_9,
+      timeout: Duration.seconds(180),
+      environment: { DB_SECRET_ARN: props.dbSecret.secretArn, ...props.env },
+      vpc: props.dbVpc,
+      vpcSubnets: props.subnetSelection,
+      allowPublicSubnet: true,
+      memorySize: 2048,
+    });
+
+    // Allow handler to read DB secret
+    props.dbSecret.grantRead(handler);
+
+    // Allow handler to connect to DB
+    props.dbSecurityGroup.addIngressRule(
+      handler.connections.securityGroups[0],
+      ec2.Port.tcp(5432),
+      "Allow connections from STAC Ingestor"
+    );
+
+    // Allow handler to write results back to DBƒ
+    props.table.grantWriteData(handler);
+
+    // Trigger handler from writes to DynamoDB table
+    handler.addEventSource(
+      new events.DynamoEventSource(props.table, {
+        // Read when batches reach size...
+        batchSize: 1000,
+        // ... or when window is reached.
+        maxBatchingWindow: Duration.seconds(10),
+        // Read oldest data first.
+        startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+        retryAttempts: 1,
+      })
+    );
+
+    return handler;
+  }
+
+  private buildApiEndpoint(props: {
+    handler: lambda.IFunction;
+    stage: string;
+    policy?: iam.PolicyDocument;
+    endpointConfiguration?: apigateway.EndpointConfiguration;
+  }): apigateway.LambdaRestApi {
+    return new apigateway.LambdaRestApi(
+      this,
+      `${Stack.of(this).stackName}-ingestor-api`,
+      {
+        handler: props.handler,
+        proxy: true,
+
+        cloudWatchRole: true,
+        deployOptions: { stageName: props.stage },
+        endpointExportName: `ingestor-api-${props.stage}`,
+
+        endpointConfiguration: props.endpointConfiguration,
+        policy: props.policy,
+      }
+    );
+  }
+
+  private registerSsmParameter(props: {
+    name: string;
+    value: string;
+    description: string;
+  }): ssm.IStringParameter {
+    const parameterNamespace = Stack.of(this).stackName;
+    return new ssm.StringParameter(
+      this,
+      `${props.name.replace("_", "-")}-parameter`,
+      {
+        description: props.description,
+        parameterName: `/${parameterNamespace}/${props.name}`,
+        stringValue: props.value,
+      }
+    );
+  }
+}
+
+export interface StacIngestorProps {
+  /**
+   * ARN of AWS Role used to validate access to S3 data
+   */
+  readonly dataAccessRole: iam.IRole;
+
+  /**
+   * URL of STAC API
+   */
+  readonly stacUrl: string;
+
+  /**
+   * Stage of deployment (e.g. `dev`, `prod`)
+   */
+  readonly stage: string;
+
+  /**
+   * Secret containing pgSTAC DB connection information
+   */
+  readonly stacDbSecret: secretsmanager.ISecret;
+
+  /**
+   * VPC running pgSTAC DB
+   */
+  readonly vpc: ec2.IVpc;
+
+  /**
+   * Security Group used by pgSTAC DB
+   */
+  readonly stacDbSecurityGroup: ec2.ISecurityGroup;
+
+  /**
+   * Boolean indicating whether or not pgSTAC DB is in a public subnet
+   */
+  readonly subnetSelection: ec2.SubnetSelection;
+
+  /**
+   * Environment variables to be sent to Lambda.
+   */
+  readonly apiEnv: Record<string, string>;
+
+  /**
+   * API Endpoint Configuration, useful for creating private APIs.
+   */
+  readonly apiEndpointConfiguration?: apigateway.EndpointConfiguration;
+
+  /**
+   * API Policy Document, useful for creating private APIs.
+   */
+  readonly apiPolicy?: iam.PolicyDocument;
+}

lib/ingestor-api/runtime/requirements.txt

@@ -0,0 +1,14 @@
+Authlib==1.0.1
+cachetools==5.1.0
+fastapi>=0.75.1
+mangum>=0.15.0
+orjson>=3.6.8
+psycopg[binary,pool]>=3.0.15
+pydantic_ssm_settings>=0.2.0
+pydantic>=1.9.0
+# Waiting for https://github.com/stac-utils/pgstac/pull/135
+# pypgstac==0.6.6
+pypgstac @ git+https://github.com/stac-utils/pgstac.git@main#egg=pygstac&subdirectory=pypgstac
+requests>=2.27.1
+# Waiting for https://github.com/stac-utils/stac-pydantic/pull/116
+stac-pydantic @ git+https://github.com/alukach/stac-pydantic.git@patch-1

lib/ingestor-api/runtime/src/__init__.py

@@ -0,0 +1,43 @@
+import os
+from getpass import getuser
+from typing import Optional
+
+from pydantic import BaseSettings, Field, AnyHttpUrl, constr
+from pydantic_ssm_settings import AwsSsmSourceConfig
+
+
+AwsArn = constr(regex=r"^arn:aws:iam::\d{12}:role/.+")
+
+
+class Settings(BaseSettings):
+    dynamodb_table: str
+
+    root_path: Optional[str] = Field(description="Path from where to serve this URL.")
+
+    jwks_url: Optional[AnyHttpUrl] = Field(
+        description="URL of JWKS, e.g. https://cognito-idp.{region}.amazonaws.com/{userpool_id}/.well-known/jwks.json"  # noqa
+    )
+
+    stac_url: AnyHttpUrl = Field(description="URL of STAC API")
+
+    data_access_role: AwsArn = Field(
+        description="ARN of AWS Role used to validate access to S3 data"
+    )
+
+    class Config(AwsSsmSourceConfig):
+        env_file = ".env"
+
+    @classmethod
+    def from_ssm(cls, stack: str):
+        return cls(_secrets_dir=f"/{stack}")
+
+
+settings = (
+    Settings()
+    if os.environ.get("NO_PYDANTIC_SSM_SETTINGS")
+    else Settings.from_ssm(
+        stack=os.environ.get(
+            "STACK", f"veda-stac-ingestion-system-{os.environ.get('STAGE', getuser())}"
+        ),
+    )
+)