# Orchard Simple Workflow Feed Ingestion General python workflow for data feed ingestion ## Other docs * [Vulnerabilities fix](VULNERABILITIES_FIX.md) * [Docker compose for dev](DOCKER_COMPOSE.md) * [Backfill tool](BACKLOAD.md) ## Installation See below for steps to setup swf-feed-ingestion to run. A video workshop of setting up SWF to run locally for development purposes can be viewed [here](https://drive.google.com/open?id=1Bx_-O6L52qaKCPLQjSUOs7Gt3JjyOHx3) ### Dependencies - Python >=3.11 We define python version in `pyproject.toml` and in base image of [Dockerfile](Dockerfile) It's possible to manage several independent python versions using `uv`. Please check [uv](https://docs.astral.sh/uv/guides/install-python/) ```shell script $ uv python install 3.12 $ uv python list $ uv python find ``` ### Fork original repository and create local clone Fork [https://github.com/theorchard/swf-feed-ingestion](https://github.com/theorchard/swf-feed-ingestion) into your own Github account ```shell script $ mkdir swf-feed-ingestion[-dev] && cd swf-feed-ingestion[-dev] $ git clone git@github.com:YOUR_GITHUB_USERNAME/swf-feed-ingestion.git . ``` Also, we can use [gh](https://cli.github.com/) GitHub CLI tool. ## Setting up local development environment There are options: * use docker compose * use uv as a virtual environment manager * use legacy pipenv package manager ### Setting up using Docker Compose Refer to [DOCKER_COMPOSE.md](DOCKER_COMPOSE.md) ### Setting up Python virtual environment: uv #### Install required libraries *Note:* To see executed commands and keys check out `Makefile` internals. ##### For development ```shell script $ make install_from_lock_include_dev ``` ##### For production ```shell script $ make install_from_lock ``` #### Some useful uv commands Note: please follow [DEPENDENCIES_MANAGEMENT.md](DEPENDENCIES_MANAGEMENT.md) guidelines when adding or updating dependent python packages. Force re-resolve and lock all dependencies: ``` uv lock --upgrade ``` Upgrade only one package and its dependencies: ``` uv lock --upgrade-package snowflake_connector_etl ``` Add a new dev package: ``` uv add --dev freezegun ``` Install python: ``` uv python install 3.11 ``` See dependencies tree: ``` uv tree ``` #### Configure `.env` file Template `.env.shadow` file contains all environment variables needed for successful connectivity, authentication, etc. Copy this file to `.env` and add values to required variables. **Note:** If a `.env` file is present in your project, you can use `$ uv run --env-file .env ` to read values from it. ```shell script $ cp .env.shadow .env ``` ### Using legacy pipenv package manager If for any reason you want to work with legacy pipenv package manager, you can use the following commands: #### pipenv in Docker Compose Set in `.env`: ``` DOCKERFILE=pipenv.Dockerfile ``` Now all services from [docker-compose.yml](docker-compose.yml) will use pipenv.Dockerfile instead of default Dockerfile. #### pipenv in local development environment First, you need to rename pipenv files to their default names: ```shell script $ cp Pipfile.off Pipfile $ cp Pipfile.off.lock Pipfile.lock ``` Now you can: * sync from lock: ```shell $ pipenv --rm sync --dev --python 3.11 $ pipenv run pip install -e . ``` * update lock of packages: ```shell $ pipenv lock --python 3.11 ``` * run unit tests: ```shell $ PIPENV_DONT_LOAD_ENV=1 pipenv run py.test tests/ ``` ## Manage AWS credentials Development SWF domain must be assigned to `SWF_DOMAIN=` env variable in the `.env` file. Next generate AWS keys/tokens using `awsume`. Make sure you have `awsume` [installed and configured](https://www.notion.so/AWS-Access-f841b9dd815d4443a80e96a86c92cd2f) Create role-based profile `dev`. Your .aws/config should look like this: ``` [profile dev] source_profile = prod role_session_name = role_arn = arn:aws:iam::103233932089:role/generic-engineer-role ``` To generate temp keys for `dev` AWS account run the following command. Note: It will ask you for MFA code of `prod` AWS account ```bash awsume dev ``` OK, you have temporary dev AWS creds in environment variables of current session. Now it's time to put them in .env file. __Important!!!__ Session based credentials valid for __1 HOUR__ only! Find AWS-related environmental variables ```bash set | grep AWS_ ``` 4. copy variables values listed below to .env (replacing existing ones) ``` AWS_SECRET_ACCESS_KEY= AWS_SESSION_TOKEN= AWS_ACCESS_KEY_ID= ``` 5. add AWS_SECURITY_TOKEN with the same value as AWS_SESSION_TOKEN (not needed anymore, as dev.py has support for boto2 compatibility) Additionally **PROD AWS** keys may be added to the following env variables. They are used by stage loaders to load data from s3 location (like from `s3://cucumbers`). Do it ONLY if need to use different AWS account for load data from. ``` AWS_WORKFLOW_ACCESS_KEY_ID= AWS_WORKFLOW_SECRET_ACCESS_KEY= AWS_WORKFLOW_SESSION_TOKEN= ``` ## Manage Snowflake credentials Snowflake data warehouse requires these env variables to operate correctly ```shell script SNOWFLAKE_USER= SNOWFLAKE_PASSWORD= SNOWFLAKE_DATABASE= SNOWFLAKE_SCHEMA= ``` The rest of the env variables should be left as they are in the `.env.shadow` #### Snowflake authentication using SSH-key To authenticate using SSH-key following relevant [Snowflake documentation](https://docs.snowflake.net/manuals/user-guide/snowsql-start.html#using-key-pair-authentication) TL;DR ```shell $ mkdir ~/.ssh/snowflake && cd ~/.ssh/snowflake # recommended $ openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 $ openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub ``` Then send newly generated public key to [systems@theorchard.com](mailto:systems@theorchard.com) and create **SYS** Jira ticket (type: AccessRequest). *Note:* if you are not going to put ssh-key into `~/.ssh/snowflake` you should add its location to the `SNOWFLAKE_PRIVATE_KEY_PATH` variable. ### Setting Up Snowflake Dev Database Schema Before running a flow for the first time one should create tables required for ETL flow being executed. First loging to your [Snowflake account](https://orchard.snowflakecomputing.com), change password and setup 2FA authentication To run ETL locally create your own schema in `DEV_ENGINEERING.` using SQL commands from file [docs/fact_analytics_dev_db.sql](docs/fact_analytics_dev_db.sql) file manually entering them to the `Worksheet` tab inside your account. After that create `STAGING_RAW_{flow_name}` table for a flow to be executed ```sql CREATE TABLE IF NOT EXISTS STAGING_RAW_{flow_name} LIKE FACTS.PROD.STAGING_RAW_{flow_name}; ``` *Note:* Currently this file countains tables needed by Deezer flow. You may need to add additional `staging_raw_{flow_name}` tables for other workflows where `flow_name` is the directory name inside `./feed_ingestion/flows/` of this repository ### Running the decider, worker and executor scripts: Run the scripts in this exact order (change `context_date` and any other context as required). Python environment is automatically activated by `pipenv run` ```shell $ pipenv run garcon decider {flow_name} $ pipenv run garcon worker {flow_name} $ pipenv run garcon exec {flow_name} -c "{\"context_date\": \"2021-02-01\"}" ``` *Note:* Where `flow_name` is the directory name inside `./feed_ingestion/flows/` of this repository. ## How workflow is executed in ECS environment using Dockerfile entrypoint and can be emulated locally. Docker image is built using Dockerfile and entrypoint is set to `./entrypoint.sh`. #### Overview The entrypoint.sh script is designed to be the entry point for a Docker container that uses Supervisord to manage processes. It ensures that the necessary environment variables are set and starts Supervisord with the appropriate configuration file. #### Prerequisites Environment variables: ``` GARCON_FLOW_NAME: Name of the Garcon flow to be used. CUSTOM_SUPERVISOR_CONF (optional): Path to a custom Supervisord configuration file. Defaults to conf/supervisor.conf if not set. ``` #### Usage Build the Docker Image: ``` docker build -t your-image-name . ``` Run the Docker Container: Use the docker run command to start the container. Make sure to pass the all other required environment variables. ``` docker run -e GARCON_FLOW_NAME=your_flow_name -e CUSTOM_SUPERVISOR_CONF=conf/your_custom_supervisor.conf your-image-name ``` If you do not have a custom Supervisord configuration file, you can omit the CUSTOM_SUPERVISOR_CONF environment variable: Custom supervisor configuration can be used if you want to define custom settings like more workers, decider, and numprocs. Example: CUSTOM_SUPERVISOR_CONF=conf/beatport_supervisor.conf ``` docker run -e GARCON_FLOW_NAME=your_flow_name your-image-name ``` ## How Garcon/SWF works This project works on top of Garcon and Amazon SWF. To understand more about Garcon, please visit: [Garcon](https://github.com/xethorn/garcon). In general, this project provides a simple way for developers to create data ingestion workflow by implementing specific small tasks and linking them together. A context dictionary is passed along to feed input data for each task. All you need to create for each new workflow to get started: flow.py, tasks.py and the .yml config file for this flow. ### Hello World Garcon ~~A good introduction to Garcon can be found in the [Hello World Garcon project](https://github.com/someboredkiddo/hello_world_garcon). It is recommended that one set-ups and successfully runs this Workflow bef ore beginning development on swf-feed-ingestion.~~ This project needs some maintainance to bring boto up to v3 and make it work, so skip this for now! ### garcon-activity-local The `garcon-activity-local` command's purpose is to make it easy to analyze and run a garcon workflow's activities in one's local dev environment, without going through the hassle of setting up deciders/workers, waiting for SWF, etc. Details on how to install and use `garcon-activity-local` can be found in [garcon-contrib/contrib-cli](https://github.com/theorchard/garcon-contrib/tree/master/contrib-cli) Example usage within `swf-feed-ingestion`: List of Activities in `apple_music_streams` that can be run locally ``` $ feed_ingestion.flows.apple_music_streams.flow -c Flow list ``` Run Apple Music Stream's [reporter_to_s3](https://github.com/theorchard/swf-feed-ingestion/blob/master/feed_ingestion/flows/apple_music_streams/flow.py#L115) Activity locally with a context as defined in `am_streams.json` ``` $ feed_ingestion.flows.apple_music_streams.flow -c Flow run apple_music_streams_feed_ingestion_reporter_to_s3 -cf ./am_streams.json ``` ## swf-feed-ingestion Application ### Generic Workflow Pattern In general, the ingestion workflows have three main steps: 1. Download raw files from the partner - Grab all raw stream and metadata files and save them to our s3://cucumbers bucket 2. Load `staging_raw_{flow_name}` for selected workflow - Raw data for a given flow should be de-normalized (i.e. files should be joined together) and loaded into a single table 3. Load data into `fact_analytics` and `fact_analytics_error` tables - Analytics from the `staging_raw_{flow_name}` table has been joined with the appropriate releases/tracks in our dimension tables and loaded into our `fact` and `fact_error` tables. A detailed end-to-end diagram of above steps can be seen here: [https://www.lucidchart.com/documents/edit/75d822d7-2a51-4118-a6fe-cfca0b397374](https://www.lucidchart.com/documents/edit/75d822d7-2a51-4118-a6fe-cfca0b397374) ### Folder structure ``` . ├── conf (1) ├── ... ├── feed_ingestion │   ├── bin │   ├── common │   ├── conf (2) │   ├── flows (3) │ │   ├── apple_music │ │ │   └── config.py (7) │ │   └── itunes │   ├── common (8) │   ├── conf (2) │   ├── tasks (4) │   └── util (5) ├── ... └── tests (6) └── ... ``` 1. `conf` - Contains supervisord configuration files 2. `conf` - Contains global YAML configuration files and some legacy YAML configs for existing workflows. At present local workflow config variables should always be defined in flow's `config.py` file (7) 3. `feed_ingestion` - Contains all the work flows. If you need to add a new workflow, create a package here and add the name of your package to the `__all__` list in `__init__.py` 4. `tasks` - Generic tasks that are being used among all workflows 5. `util` - Generic utility methods being used among all workflows 6. `tests` - unit tests. 7. `config.py` - flow specific configs 8. `common` - common modules shared among all the workflows ### Cucumber (a workflow's working/raw files in s3) Working/raw s3 files for a workflow should be stored in `s3://cucumbers/` (and `s3://dev-cucumbers` for dev) *Note:* this has nothing to do with the cucumbers testing application ... it is called cucumbers because it has raw files in it (sad trombone) ``` s3://[dev-]cucumbers ├── (1) │   ├── archives (2) │ │   ├── (3) │ │   ├── │ │   ├── │ │   └── ... │   ├── temp_staging_raw_{flow_name} (4) │ │   ├── (3) │ │   ├── │ │   ├── │ │   └── ... │   ├── staging_raw_{flow_name} (5) │ │   ├── (3) │ │   ├── │ │   ├── │ │   └── ... │   ├── fact_analytics (6) │ │   ├── (3) │ │   ├── │ │   ├── │ │   └── ... │   └── fact_analytics_error (7) │    ├── (3) │    ├── │    ├── ... │    └── ├── ... └── (1) ``` 1. `` - Name of the flow in upper camel case (i.e. `playstation_video` is `PlaystationVideo`) 2. `archives` - location where past raw data files processed by a workflow can be found 3. `` - the date when the workflow was executed 4. `temp_staging_raw_{flow_name}` - A snapshot\* of the feed's data which goes into the temporary staging table that will be used to populate its `staging_raw` table (say that 5 times fast) 5. `staging_raw_{flow_name}` - A snapshot\* of the feed's data which goes into the feed's `staging_raw_{flow_name}` table (i.e. `staging_raw_itunes`) 6. `fact_analytics` - Snapshot\* of the feed's data going into the `fact_analytics` table 7. `fact_analytics_error` - A snapshot\* of the feed's data going into the `fact_analytics_error` table A workflow can add additional subkeys when needed, but should document them and try to add them in a way that makes sense for other workflows to use them. ### Drop Bucket Location where new, unprocessed source files for a workflow can be put and grabbed by the Workflow to be processed. After a workflow has processed a file in the drop location it should be removed and placed in the `archives/` subkey Location for production is: `s3://prod-orcdbucket/feed-drop//` Location for dev/qa is: `s3://dev-feed-drop//` The `` should be in upper camel case and match name inside `s3://[dev-]cucumbers` ### Coding Conventions 1. All file and folder names are lowercase with `_` as word separator 2. Workflow specific tasks must reside in `/feed_ingestion/flows/{particular flow}/` 3. Avoid having a context dictionary that is too large. Load configuration from config files within tasks if possible. This will simplify the code and unit tests a lot. 4. Only abstract out utility methods/module/class if it is a very generic task. Such task should be very small and accept very small number of input params. 5. Unless a compelling reason exists (ex. piping between processes to avoid writing to disk), try and keep code in Python and use boto instead of sub processes. This makes it easier to unit test the code. 6. All workflows should be using python configs (Practice of adding YAML configs considered deprecated since 10/12/2015, though the existing old ones should still be used and not duplicated) ### Support for optional and required contexts Now it is implemented for Apple Music streams, itunes. It requires code changes to add optional contexts support to other flows. Configuration allows to specify either required or options contexts. Example of configuration: ``` contexts_config = { # licensor, e.g. default, theorchard, sme 'default': { # report, e.g. default, amContainer, amStreams 'default': { 'required': [ '80026921', '80028967', '85520934' ] }, 'amContainer': { 'optional': [ '80026921', '80028967' ] } }, 'sme': { ... }, 'theorchard': { ... } } ``` If licensor has value 'default', it is applied to all licensors. There could be multiple licensors. If report has value 'default', it is applied to all reports. There could be multiple reports. Please, find more details [here](https://www.notion.so/Data-ingestion-support-for-required-and-optional-contexts-866c62d2a4134efa9d2e8ed8a19b1460#81849956b42741bfa85438017e3b3880) ## Logging ### Exception handling in Sentry: Python exceptions generated by each flow can be be accessed in [Sentry.io](https://sentry.io/organizations/the-orchard/issues/) ### Application logs Datadog: Application (worker side) logs are accamulated in AWS CloudWatch get forwarde to [Datadog](https://app.datadoghq.com/logs?cols=%40aws.awslogs.logGroup&event&from_ts=1621334385808&index=&integration_id=&integration_short_name=&live=true&messageDisplay=inline&query=service%3Acloudwatch+%40aws.awslogs.logGroup%3A%2Aprod%2Aswf%2A&saved_view=83722&stream_sort=desc&to_ts=1621337985808&viz=stream) Follow this steps to access them: 1. Open Datadog 'Logs' -> '[Search](https://app.datadoghq.com/logs)' panel 2. In the searchbar enter `service:cloudwatch` and `@aws.awslogs.logGroup:*prod*swf*` with the space between them ('\*' is a wildcard) 3. Enter additional keywords to the searchbar to narrow donw your search ### Failing SWF workflows in Datadog Failing SWF workflows also can be found in Datadog. To do this follow this simple steps: 1. Open Datadog 'Monitors' -> '[Triggered Monitors](https://app.datadoghq.com/monitors/triggered)' panel 2. Mark `SWF Workflow {{workflowtypename.name}} failed` checkbox in Monitor sub-menu ### Workflow Conventions 1. Name of the workflow must be the same as the name of the folder inside `./feed_ingestion/flows/` in lowercase with `_` as the word separator. 2. Workflows should be idempotent (which means, has no additional effect if it is called more than once) by date. For example, running the same feed workflow for the same date should not result in re-processing the same files and duplicate data being loaded into `raw` or `fact` tables. 3. A workflow for an already loaded day should not achieve idempotency by backing out the data and loading it again (i.e. to avoid unnecessary load or lack of analytics on snowflake when the workflow is rerunning). Instead it should check their overall status in [Feed Status](README.md#feed-status). 4. Partially run workflows should resume from where they left off and not rerun already completed tasks/activities. One can use the [Feed Status](README.md#feed-status) module to handle this. 5. Email notifications should be sent out at the end of each workflow. The email message should show as much information as possible regarding to how much data was ingested into the raw table, how much data can successfully be matched and ingested into fact table, the percentage of error, etc. Anything that can improve the transparency of the ETL. The itunes workflow is a good example of this. 1. All production workflows currently sending SNS message to the topic: *prod-swf-feed-ingestion* 2. For development environment, please send message to: *dev-feed-ingestion* ### Workflow Tools & Patterns #### garcon-contrib Re-usable Garcon tasks reside in [garcon-contrib](https://github.com/theorchard/garcon-contrib). This library includes generic tasks for UNLOADing/COPYing data into [Snowflake](https://github.com/theorchard/garcon-contrib/tree/master/contrib-snowflake), [SFTP](https://github.com/theorchard/garcon-contrib/tree/master/contrib-sftp) downloads/uploads, [s3 manipulation](https://github.com/theorchard/garcon-contrib/blob/master/contrib-aws/garcon/contrib/aws/s3.py), etc. Always check [garcon-contrib](https://github.com/theorchard/garcon-contrib) to see if a task has been implemented before writing a new one. #### Feed Status All workflows should update the feed status for a particular day's run 1. Do not use the `advance_overall_status` Garcon Activity, instead use the `feed_status.set_overall_status` method directly in your Activity Task. This will avoid accidental advancing of a feed's overall status. 2. All workflows should update the overall status feed and include the following `SOURCE_NOT_AVAIL`, `SOURCE_AVAILABLE`, `SOURCE_DOWNLOADED`, `POPULATED_RAW_TABLE`, `INGESTED`. This will be useful for feed troubleshooting.