---
pipeline:
  name: pqs1_data_source_kafka
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/bei/neon-avro-kafka-loader:v3.3.0
  image_pull_secrets:
  - battelleecology-quay-read-all-pull-secret
  env:
    OUT_PATH: /pfs/out
    SOURCE_TYPE: "pqs1"
    LOGLEVEL: debug
    YEAR_INDEX: "5"
    MONTH_INDEX: "6"
    DAY_INDEX: "7"
    KAFKA_RETENTION_DAYS: "15"
  secrets:
  - name: pachyderm-kafka-auth
    env_var: KAFKA_USER
    key: KAFKA_USER
  - name: pachyderm-kafka-auth
    env_var: KAFKA_PASSWORD
    key: KAFKA_PASSWORD
  - name: pachyderm-kafka-auth
    env_var: KAFKA_BROKER
    key: KAFKA_BROKER
  - name: l0-bucket
    env_var: BUCKET_NAME
    key: LO_BUCKET
  cmd:
  - sh
  - "-c"
  - |-
    /bin/bash <<'EOF'
    # Get GCP zone
    set -euo pipefail
    IFS=$'\n\t'
    meta=$(curl -sH "Metadata-Flavor: Google" "http://metadata/computeMetadata/v1/instance/zone")
    zone=$(echo $meta | cut -d "/" -f 4)
    echo $zone
    # 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)
    # Get date from input path. Terminal path structure must be /SOURCE_TYPE/YYYY/MM/DD/SITE_FILE
    # Datum must be set at /SOURCE_TYPE/YYYY/MM/DD or /SOURCE_TYPE/YYYY/MM/DD/SITE_FILE
    date_path=$(echo $import_trigger | cut -f $YEAR_INDEX,$MONTH_INDEX,$DAY_INDEX -d "/")
    echo $date_path
    date_str=$(date -u +%Y-%m-%d -d $date_path)
    # Get each site to run
    if [[ -f ${import_trigger} ]]; then
      import_trigger_glob="${import_trigger}"
    else
      import_trigger_glob="${import_trigger}/*"
    fi
    for site_kafka in $import_trigger_glob; do
      site_file=$(basename $site_kafka) # Strip off any path prefix
      site=$(echo $site_file | cut -f 1 -d "." --only-delimited) # Extract the site from site.kafka. Ignore site-only files (e.g. CPER vs. CPER.kafka)
      type=$(echo $site_file | cut -f 2 -d "." --only-delimited) # Extract the 'kafka' from site.kafka
      if [ "$type" != "kafka" ]
      then
        echo "$site_file is not indicated to be streaming from Kafka. Skipping..."
        continue
      elif [ "$(date -u +%s -d "$date_str")" -lt "$(date -u +%s -d "$kafka_min_date")" ]
      then
        echo -n "Cannot extract $date_str Kafka data for $site. "
        echo -n "Today's date ($date_today) is beyond the Kafka retention period ($KAFKA_RETENTION_DAYS days). Skipping..."
        continue
      fi
      # We are ok to run
      echo "Extracting $date_str kafka data for $site"
      ./extract-kafka-sensor.py -s $site -S $SOURCE_TYPE -D "$OUT_PATH/$SOURCE_TYPE" -d $date_str --only current --consumer client.rack=$zone
      date_str_1=$(date +%Y-%m-%d -d "$date_str + 1 day")
      ./extract-kafka-sensor.py -s $site -S $SOURCE_TYPE -D "$OUT_PATH/$SOURCE_TYPE" -d $date_str_1 --only noncurrent --consumer client.rack=$zone
    done
    if [[ -d "$OUT_PATH/$SOURCE_TYPE" ]]; then
      linkdir=$(mktemp -d)
      shopt -s globstar
      out_parquet_glob="${OUT_PATH}/**/*.parquet"
      # /pfs/out/pqs1/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
input:
  pfs:
    name: import_trigger
    repo: pqs1_cron_daily_and_date_control
    # Must be datum by day (e.g. /SOURCE_TYPE/*/*/*) or by day/site (e.g. /SOURCE_TYPE/*/*/*/*)
    glob: "/pqs1/*/*/*"
parallelism_spec:
  constant: 1
autoscaling: true
resource_requests:
  memory: 300M
  cpu: 1.3
resource_limits:
  memory: 1.5G
  cpu: 2
sidecar_resource_requests:
  memory: 2G
  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"
    }  
  ] }
