---
pipeline:
  name: pqs1_data_source_trino
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/neonscience/neon-is-data-src-trino:v1.1.0
  cmd:
  - sh
  - "-c"
  - |-
    /bin/bash <<'EOF'
    # Use bash-scrict mode. See http://redsymbol.net/articles/unofficial-bash-strict-mode/
    set -euo pipefail
    IFS=$'\n\t'
    # Refresh interim directories with each datum (otherwise they persist and cause probs)
    interimDir="/tmp/interimData"
    rm -rf $interimDir
    # Get today's date for evaluating kafka data retention period
    date_today=$(date -u +%Y-%m-%d)
    kafka_min_date=$(date -u -d "$KAFKA_RETENTION_DAYS days ago" +%Y-%m-%d)
    # Run first module - data_source_site (pull data from database by site)
    # Split data source path
    for path in $(find -L $import_trigger -type f); do
      echo "Processing $path"
      p=${path#/pfs}
      IFS="/"; arr=($p); unset IFS;
      year=${arr[3]}
      month=${arr[4]}
      day=${arr[5]}
      site=${arr[6]}
      type=$(echo $site | cut -f 2 -d "." --only-delimited); # Extract the "kafka" from site.kafka if present
      if [ "$type" = "kafka" ] && [ $(date -u +%s -d $year-$month-$day) -lt $(date -u +%s -d $kafka_min_date) ]
      then
        site=$(echo $site | cut -f 1 -d "." --only-delimited); # Extract the site from site.kafka.
        echo "$year/$month/$day for $site is indicated to be streaming from Kafka but has passed the Kafka retention period ($KAFKA_RETENTION_DAYS days)."
      elif [ "$type" = "kafka" ]
      then
        echo "$year/$month/$day/$site is indicated to be streaming from Kafka. Skipping..."
        continue
      fi
      echo "Extracting data from Trino for $year/$month/$day/$site"
      export GEN_DATE=$year-$month-$day
      export GEN_SITE_NAME=$site
      export GEN_OUTPUT_DIR=$interimDir/$SOURCE_TYPE/$year/$month/$day
      export REQUESTS_CA_BUNDLE=/etc/pki/tls/cert.pem
      mkdir -p $GEN_OUTPUT_DIR
      /usr/src/app/genscript/genparquet.py --storesitename --codec gzip --truncperiod second
    done
    # Run second module - parquet_linkmerge (merges data from a source id that collected data from multiple sites in one day
    python3 -m parquet_linkmerge.parquet_linkmerge_main
    # Export L0 data to bucket
    if [[ -d "$OUT_PATH/$SOURCE_TYPE" ]]; then
      linkdir=$(mktemp -d)
      shopt -s globstar
      out_parquet_glob="${OUT_PATH}/**/*.parquet"
      # Example: /pfs/out/li191r/2023/01/01/12345/data/file.parquet
      echo "Linking output files to ${linkdir}"
      set -x
      for f in $out_parquet_glob; do
        # Parse the path
        [[ "$f" =~ ^$OUT_PATH/(.*)/([0-9]+)/([0-9]+)/([0-9]+)/(.*)/data/(.*)$ ]]
        fsourcetype="${BASH_REMATCH[1]}"
        fyear="${BASH_REMATCH[2]}"
        fmonth="${BASH_REMATCH[3]}"
        fday="${BASH_REMATCH[4]}"
        fsourceid="${BASH_REMATCH[5]}"
        fname="${BASH_REMATCH[6]}"
        outdir="${linkdir}/v1/${fsourcetype}/ms=${fyear}-${fmonth}/source_id=${fsourceid}"
        mkdir -p "${outdir}"
        ln -s "${f}" "${outdir}/${fname}"
      done
      set +x
      echo "Syncing files to bucket"
      rclone \
        --no-check-dest \
        --copy-links \
        --gcs-bucket-policy-only \
        --gcs-no-check-bucket \
        copy \
        "${linkdir}" \
        ":gcs://${BUCKET_NAME}"
      echo "Removing temporary files"
      rm -rf $linkdir
    fi
    EOF
  env:
    # Environment variables for data conversion step
    GEN_YAML_CONF: "/usr/src/app/genscript/configs/pqs1_streams.yaml"
    GEN_SCHEMA_FILE: "/usr/src/app/schemas/pqs1/pqs1.avsc"
    LOG_LEVEL: INFO
    REQUESTS_CA_BUNDLE: "/etc/pki/tls/cert.pem"
    # Environment variables for linkmerge step
    IN_PATH: /tmp/interimData
    OUT_PATH: /pfs/out
    SOURCE_TYPE_INDEX: '3'
    YEAR_INDEX: '4'
    MONTH_INDEX: '5'
    DAY_INDEX: '6'
    SOURCE_ID_INDEX: '7'
    KAFKA_RETENTION_DAYS: "15"
    # Environment variables for bash code
    SOURCE_TYPE: 'pqs1'
  secrets:
  - name: pachd-trino-secret
    key: TRINO_HOST
    env_var: PRESTO_HOST
  - name: pachd-trino-secret
    key: TRINO_PASSWORD
    env_var: PRESTO_PASSWORD
  - name: pachd-trino-secret
    key: TRINO_USER
    env_var: PRESTO_USER
  - name: l0-bucket
    env_var: BUCKET_NAME
    key: LO_BUCKET
input:
  pfs:
    name: import_trigger
    repo: pqs1_cron_daily_and_date_control
    glob: "/pqs1/*/*/*"
output_branch: master
parallelism_spec:
  constant: 1
autoscaling: true
resource_requests:
  memory: 400M
  cpu: 1.2
resource_limits:
  memory: 800M
  cpu: 2
sidecar_resource_requests:
  memory: 3G
  cpu: 0.5
datum_set_spec:
  number: 1
scheduling_spec:
  node_selector:
    cloud.google.com/gke-ephemeral-storage-local-ssd: "true"
    nodepool.neonscience.org/pipeline: "yes"
    cloud.google.com/gke-spot: "true"
pod_spec: |-
  { "tolerations": [
    {
      "key": "nodepool.neonscience.org/pipeline",
      "operator": "Exists"
    },
    {
      "effect": "NoSchedule",
      "key": "cloud.google.com/gke-spot",
      "operator": "Exists"
    }  
  ] }
