Version/aeolus

.github/workflows/deploy.yml

@@ -0,0 +1,76 @@
+name: Deploy ingestor
+
+on:
+  push:
+    branches:
+      - version/aeolus
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repo
+        uses: actions/checkout@v4
+
+      - name: Configure aws credentials
+        uses: aws-actions/configure-aws-credentials@master
+        with:
+          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_PROD }}
+          aws-secret-access-key: ${{ secrets.AWS_SECRET_KEY_PROD }}
+          aws-region: ${{ secrets.AWS_REGION }}
+
+      - name: Get envionmental values
+        uses: aws-actions/aws-secretsmanager-get-secrets@v2
+        with:
+          secret-ids: |
+            AEOLUS, openaq-env/aeolus
+          name-transformation: uppercase
+          parse-json-secrets: true
+
+      - uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+
+
+      - name: Install CDK
+        run: |
+          npm install -g aws-cdk
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+
+      - name: Install Poetry
+        uses: snok/install-poetry@v1
+
+      - name: Deploy stack
+        env:
+          ENV: "aeolus"
+          PROJECT: "openaq"
+
+          ## deployment variables
+          # CDK_ACCOUNT: ${{ secrets.CDK_ACCOUNT }}
+          # CDK_REGION: ${{ secrets.CDK_REGION }}
+
+          VPC_ID: ${{ env.AEOLUS_VPC_ID }}
+
+          TOPIC_ARN: ${{ env.AEOLUS_FETCH_OBJECT_TOPIC_ARN }}
+
+          ## application variables
+          DATABASE_READ_USER: ${{ env.AEOLUS_DATABASE_READ_USER }}
+          DATABASE_READ_PASSWORD: ${{ env.AEOLUS_DATABASE_READ_PASSWORD }}
+          DATABASE_WRITE_USER: ${{ env.AEOLUS_DATABASE_WRITE_USER }}
+          DATABASE_WRITE_PASSWORD: ${{ env.AEOLUS_DATABASE_WRITE_PASSWORD }}
+          DATABASE_DB: ${{ env.AEOLUS_DATABASE_DB }}
+          DATABASE_HOST: ${{ env.AEOLUS_DATABASE_HOST }}
+          DATABASE_PORT: ${{ env.AEOLUS_DATABASE_PORT }}
+          FETCH_BUCKET: ${{ env.AEOLUS_FETCH_BUCKET }}
+          ETL_BUCKET: ${{ env.AEOLUS_FETCH_BUCKET }}
+          PAUSE_INGESTING: False
+
+
+        working-directory: ./cdk
+        run: |
+          poetry self add poetry-plugin-export
+          poetry install
+          cdk deploy openaq-ingest-aeolus --require-approval never

README.md

@@ -2,3 +2,5 @@
 
 
 # Testing a realtime file
+
+# Testing files

benchmark.py

@@ -0,0 +1,95 @@
+import logging
+import os
+import sys
+import argparse
+from time import time
+import re
+
+logger = logging.getLogger(__name__)
+
+parser = argparse.ArgumentParser(
+    description="""
+Test benchmarks for ingestion
+    """)
+
+parser.add_argument(
+	'--name',
+	type=str,
+	required=False,
+	default="4xlarge",
+	help='Name to use for the test'
+	)
+parser.add_argument(
+	'--env',
+	type=str,
+	default='.env',
+	required=False,
+	help='The dot env file to use'
+	)
+parser.add_argument(
+	'--debug',
+	action="store_true",
+	help='Output at DEBUG level'
+	)
+args = parser.parse_args()
+
+if 'DOTENV' not in os.environ.keys() and args.env is not None:
+    os.environ['DOTENV'] = args.env
+
+if args.debug:
+    os.environ['LOG_LEVEL'] = 'DEBUG'
+
+from ingest.settings import settings
+from fake import config, get_locations, as_realtime
+from ingest.fetch import load_realtime
+
+logging.basicConfig(
+    format='[%(asctime)s] %(levelname)s [%(name)s:%(lineno)s] %(message)s',
+    level=settings.LOG_LEVEL.upper(),
+    force=True,
+)
+
+f = open(f"benchmark_ingest_output_{args.name}.csv", "w")
+f.writelines("name,key,locations,inserted_nodes,updated_nodes,total_meas,inserted_meas,ingest_time,process_time,log_time,copy_time,load_process_time\n")
+n = 10
+locations = [50, 250, 1000]
+keys = []
+ii = 1
+
+## make a set of files
+for r in locations:
+	for i in range(n):
+		config(source=f"benchmark-test-{r}-{i+1}", gz=True)
+		l = get_locations(n=r)
+		key = as_realtime(l["locations"], l["latitude"], l["longitude"])
+		keys.append({ "key": key, "locations": len(l["locations"]) })
+		ii=+1
+
+
+## ingest each of the
+for i, k in enumerate(keys):
+	key = k["key"]
+	locations = k["locations"]
+	logger.info(f"Ingesting {i+1} of {len(keys)}: {key} with {locations} locations")
+
+	start_time = time()
+	copy_time, load_process_time, log_time, notice = load_realtime([
+		(-1, key, None)
+	  ])
+	m = re.findall('([a-z-]+): (.+?),', notice)
+
+	process_time = round(float(m[17][1]))
+	total_meas = int(m[0][1])
+	inserted_meas = int(m[9][1])
+	updated_nodes = int(m[8][1])
+	inserted_nodes = int(m[11][1])
+	ingest_time = round((time() - start_time)*1000)
+	f.writelines(f"'{args.name}','{key}',{locations},{inserted_nodes},{updated_nodes},{total_meas},{inserted_meas},{ingest_time},{process_time},{log_time},{copy_time},{load_process_time}\n")
+
+	logger.info(
+		"loaded realtime records, timer: %0.4f, process: %0.4f",
+		ingest_time, process_time
+		)
+
+
+f.close()

cdk/app.py

@@ -3,6 +3,7 @@
     Environment,
     Tags,
 )
+import os
 
 from lambda_ingest_stack import LambdaIngestStack
 
@@ -19,16 +20,23 @@
 
 app = aws_cdk.App()
 
+env = Environment(
+	account=os.environ['CDK_DEFAULT_ACCOUNT'],
+	region=os.environ['CDK_DEFAULT_REGION']
+	)
+
 ingest = LambdaIngestStack(
     app,
     f"openaq-ingest-{settings.ENV}",
     env_name=settings.ENV,
     lambda_env=lambda_env,
     fetch_bucket=settings.FETCH_BUCKET,
-    ingest_lambda_timeout=settings.INGEST_LAMBDA_TIMEOUT,
-    ingest_lambda_memory_size=settings.INGEST_LAMBDA_MEMORY_SIZE,
-    ingest_rate_minutes=settings.INGEST_RATE_MINUTES,
+	vpc_id=settings.VPC_ID,
+    lambda_timeout=settings.LAMBDA_TIMEOUT,
+    lambda_memory_size=settings.LAMBDA_MEMORY_SIZE,
+    rate_minutes=settings.RATE_MINUTES,
     topic_arn=settings.TOPIC_ARN,
+	env=env,
 )
 
 Tags.of(ingest).add("project", settings.PROJECT)

cdk/cdk.json

@@ -1,5 +1,5 @@
 {
-  "app": "python3.8 app.py",
+  "app": "poetry run python app.py",
   "context": {
     "aws-cdk:enableDiffNoFail": "true",
     "@aws-cdk/core:stackRelativeExports": "true",

cdk/config.py

@@ -1,5 +1,8 @@
 from typing import List
-from pydantic import BaseSettings
+from pydantic_settings import (
+    BaseSettings,
+    SettingsConfigDict,
+    )
 from pathlib import Path
 from os import environ
 
@@ -8,18 +11,17 @@ class Settings(BaseSettings):
     FETCH_BUCKET: str
     ENV: str = "staging"
     PROJECT: str = "openaq"
-    INGEST_LAMBDA_TIMEOUT: int = 900
-    INGEST_LAMBDA_MEMORY_SIZE: int = 1536
-    INGEST_RATE_MINUTES: int = 15
+    LAMBDA_TIMEOUT: int = 900
+    LAMBDA_MEMORY_SIZE: int = 1536
+    RATE_MINUTES: int = 15
     LOG_LEVEL: str = 'INFO'
     TOPIC_ARN: str = None
+    VPC_ID: str = None
 
-    class Config:
-        parent = Path(__file__).resolve().parent.parent
-        if 'DOTENV' in environ:
-            env_file = Path.joinpath(parent, environ['DOTENV'])
-        else:
-            env_file = Path.joinpath(parent, ".env")
+
+    model_config = SettingsConfigDict(
+        extra="ignore", env_file=f"../{environ.get('DOTENV', '.env')}", env_file_encoding="utf-8"
+    )
 
 
 settings = Settings()

cdk/lambda_ingest_stack.py

@@ -2,8 +2,10 @@
 from typing import Dict
 
 from aws_cdk import (
+	Environment,
     aws_lambda,
     aws_s3,
+	aws_ec2,
     Stack,
     Duration,
     aws_events,
@@ -24,18 +26,23 @@ def __init__(
         self,
         scope: Construct,
         id: str,
+        env: Environment,
         env_name: str,
         lambda_env: Dict,
         fetch_bucket: str,
-        ingest_lambda_timeout: int,
-        ingest_lambda_memory_size: int,
-        ingest_rate_minutes: int = 15,
+        lambda_timeout: int,
+        lambda_memory_size: int,
+        rate_minutes: int = 15,
         topic_arn: str = None,
+        vpc_id: str = None,
         **kwargs,
     ) -> None:
         """Lambda plus cronjob to ingest metadata,
         realtime and pipeline data"""
-        super().__init__(scope, id, *kwargs)
+        super().__init__(scope, id, env=env,*kwargs)
+
+        if vpc_id is not None:
+            vpc_id = aws_ec2.Vpc.from_lookup(self, f"{id}-vpc", vpc_id=vpc_id)
 
         ingest_function = aws_lambda.Function(
             self,
@@ -58,11 +65,12 @@ def __init__(
                 ],
             ),
             handler="ingest.handler.handler",
-            runtime=aws_lambda.Runtime.PYTHON_3_8,
+			vpc=vpc_id,
+            runtime=aws_lambda.Runtime.PYTHON_3_12,
             allow_public_subnet=True,
-            memory_size=ingest_lambda_memory_size,
+            memory_size=lambda_memory_size,
             environment=stringify_settings(lambda_env),
-            timeout=Duration.seconds(ingest_lambda_timeout),
+            timeout=Duration.seconds(lambda_timeout),
             layers=[
                 create_dependencies_layer(
                     self,
@@ -81,12 +89,12 @@ def __init__(
 
         # Set how often the ingester will run
         # If 0 the ingester will not run automatically
-        if ingest_rate_minutes > 0:
+        if rate_minutes > 0:
             aws_events.Rule(
                 self,
                 f"{id}-ingest-event-rule",
                 schedule=aws_events.Schedule.cron(
-                    minute=f"0/{ingest_rate_minutes}"
+                    minute=f"0/{rate_minutes}"
                 ),
                 targets=[
                     aws_events_targets.LambdaFunction(ingest_function),

cdk/utils.py

@@ -19,20 +19,22 @@ def create_dependencies_layer(
         function_name: str,
         requirements_path: Path
 ) -> aws_lambda.LayerVersion:
-    requirements_file = str(requirements_path.resolve())
+    #requirements_file = str(requirements_path.resolve())
     output_dir = f'../.build/{function_name}'
     layer_id = f'openaq-{function_name}-{env_name}-dependencies'
 
-    if not environ.get('SKIP_PIP'):
-        print(f'Building {layer_id} from {requirements_file} into {output_dir}')
+    if not environ.get('SKIP_BUILD'):
+        print(f'Building {layer_id} into {output_dir}')
         subprocess.run(
-            f"""python3.8 -m pip install -qq -r {requirements_file} \
+            f"""
+             poetry export --without=cdk -o requirements.txt --without-hashes && \
+             poetry run python -m pip install -qq -r requirements.txt \
             -t {output_dir}/python && \
             cd {output_dir}/python && \
             find . -type f -name '*.pyc' | \
               while read f; do n=$(echo $f | \
               sed 's/__pycache__\///' | \
-              sed 's/.cpython-[2-3] [0-9]//'); \
+              sed 's/.cpython-[2-3][0-9]//'); \
               cp $f $n; \
               done \
             && find . -type d -a -name '__pycache__' -print0 | xargs -0 rm -rf \
@@ -47,5 +49,5 @@ def create_dependencies_layer(
         self,
         layer_id,
         code=layer_code,
-        compatible_runtimes=[aws_lambda.Runtime.PYTHON_3_8]
+        compatible_runtimes=[aws_lambda.Runtime.PYTHON_3_12]
     )