#!/usr/bin/env bash
set -e

ENVIRONMENT="$(hostname|sed -r 's/^([^-]+)-([^-]+)-.*-([^-]+)-([a-z0-9]{2,4}+)$/\3/g')"
GCP_MIG="gcp-web-minimetrica-${ENVIRONMENT}"
GCP_BUCKET_SESSIONS="minimetrica-sessions-${ENVIRONMENT}"

cd /var/lib/minimetrica

# создаем папку с данными, если ее нет
[ -d export ] || mkdir export
# папка для сохранения копии данных для DWH и дальнейшего копирования в Google Storage
[ -d dwh_gs_export ] || mkdir dwh_gs_export

# считаем кол-во таблиц, для которых появились необработанные архивы
export_bad_count=0
# переменные для хранения ошибок при инсерте в таблицы
hits_error=""
users_params_error=""
other_errors=""

# команда для получения списка инстансов с инстанс-группы
LIST_INSTANCES_COMMAND="gcloud compute instance-groups list-instances "${GCP_MIG}" --zone=us-central1-a --format='value(NAME)' --filter='STATUS=RUNNING'"

# копирование логов с машин инстанс-группы
while IFS='' read -r IP || [[ -n "$IP" ]]; do
    echo $IP
    rsync -a -e "ssh -o StrictHostKeyChecking=no  -o UserKnownHostsFile=/dev/null -q" --bwlimit=8000 --remove-source-files "$IP".c.hadoop-151909.internal:/var/log/nginx/export/ ./export/
done < <(eval "$LIST_INSTANCES_COMMAND")

# метрика в clickhouse
if [ "$(find export -name "hits.log.gz*")" ]; then
    export_bad_count=$(( export_bad_count + 1 ))
    dwh_batch_file_name="hits_batch_$(date +"%Y-%m-%d_%H-%M-%S").gz"
    dwh_batch_file="./dwh_gs_export/$dwh_batch_file_name"
    hits_error=$( zcat export/hits.log.gz* | /usr/local/www/minimetrika.srv.local/current/metrika-processor | tee >(gzip > "$dwh_batch_file") | clickhouse-client --query "INSERT INTO metrika.hits FORMAT TSV" 2>&1 >/dev/null && rm ./export/hits.log.gz* || (rm -f "$dwh_batch_file" && gsutil -m mv ./export/hits.log.gz* gs://"${GCP_BUCKET_SESSIONS}"/export_bad/) )
    if [ -n "$hits_error" ]; then
      echo "$hits_error";
      hits_error="\nhits: ${hits_error}"
    fi
fi

if [ "$ENVIRONMENT" = "prod" ]; then
  # перенос копии данных метрики в Google Storage для DWH
  echo "Upload minimetrika data to DWH Google Storage"
  if [ "$(find dwh_gs_export -name "hits_batch*.gz")" ]; then
      export_bad_count=$(( export_bad_count - 1 ))
      gs_file_path="gs://dwh-mini-metrika/datagather/hits/$(date +"%Y-%m-%d/%H")/"
      gsutil -m mv ./dwh_gs_export/hits_batch*.gz "$gs_file_path" || true
  #    gsutil setmeta -h "Content-Type:gzip" "$gs_file_path"hits_batch*.gz
  fi
fi

# параметры пользователей в clickhouse
if [ "$(find export -name users_params.log.gz*)" ]; then
    export_bad_count=$(( export_bad_count + 1 ))
    dwh_batch_file_name="users_params_batch_$(date +"%Y-%m-%d_%H-%M-%S").gz"
    dwh_batch_file="./dwh_gs_export/$dwh_batch_file_name"
    # input_format_allow_errors_num - максимальное кол-во ошибок при обработке одной строки
    # input_format_allow_errors_ratio - максимальный процент строк, в которых возникли ошибки
    # если input_format_allow_errors_num и input_format_allow_errors_ratio превышены, то будет ошибка
    users_params_error=$( zcat export/users_params.log.gz* | /usr/local/www/minimetrika.srv.local/current/users-params-processor | tee >(gzip > "$dwh_batch_file") | clickhouse-client --query "INSERT INTO metrika.users_params FORMAT JSONEachRow" --input_format_allow_errors_num=1 --input_format_allow_errors_ratio=0.01 --input_format_skip_unknown_fields=1 2>&1 >/dev/null && rm ./export/users_params.log.gz* || (rm -f "$dwh_batch_file" && gsutil -m mv ./export/users_params.log.gz* gs://"${GCP_BUCKET_SESSIONS}"/export_bad/) )
    if [ -n "$users_params_error" ]; then
      echo "$users_params_error";
      users_params_error="\nusers_params: ${users_params_error}"
    fi
fi

# остальное в clickhouse
cd export
tables=()

# из имени архивов в /export извлекаем наименования таблиц и добавляем в массив tables
for LOGFILE in *.gz
do
  # извлекаем имя таблицы из имени архива
  table=$(sed -E 's/^([^\.]+).*/\1/' <<<"$LOGFILE")
  tables+=("$table")
done

# убираем дубли в tables, оставляем только уникальные имена таблиц
# https://stackoverflow.com/questions/13648410/how-can-i-get-unique-values-from-an-array-in-bash
tables=($(tr ' ' '\n' <<< "${tables[@]}" | sort -u | tr '\n' ' '))

# для каждой таблицы сначала находим все архивы и вставляем их батчем
for table in "${tables[@]}";
do
  if [ "$(find . -name "$table.log.gz*")" ]; then
    export_bad_count=$(( export_bad_count + 1 ))
    dwh_batch_file_name="$table"_batch_"$(date +"%Y-%m-%d_%H-%M-%S").gz"
    dwh_batch_file="../dwh_gs_export/$dwh_batch_file_name"
    # input_format_allow_errors_num - максимальное кол-во ошибок при обработке одной строки
    # input_format_allow_errors_ratio - максимальный процент строк, в которых возникли ошибки
    # если input_format_allow_errors_num и input_format_allow_errors_ratio превышены, то будет ошибка
    other_err=$( zcat "$table".log.gz* | /usr/local/www/minimetrika.srv.local/current/log-processor | tee >(gzip > "$dwh_batch_file") | clickhouse-client --query "INSERT INTO metrika.$table FORMAT JSONEachRow" --input_format_allow_errors_num=1 --input_format_allow_errors_ratio=0.01 --input_format_skip_unknown_fields=1 2>&1 >/dev/null && rm "$table".log.gz* || (rm -f "$dwh_batch_file" && gsutil -m mv -r "$table".log.gz* gs://"${GCP_BUCKET_SESSIONS}"/export_bad/) )
    if [ -n "$other_err" ]; then
      echo "$other_err";
      other_errors="${other_errors}\n${table}: ${other_err}"
    fi
  fi
done

if [ "$ENVIRONMENT" = "prod" ]; then
  # узнаем, появились ли необработанные данные на проде
  export_bad_count=$(( export_bad_count - $(find ../dwh_gs_export -name "*.gz" | wc -l) ))
  # перенос копии данных метрики в Google Storage для DWH
  echo "Upload minimetrika data to DWH Google Storage"
  if [ "$(find ../dwh_gs_export -name "*.gz")" ]; then
      gs_file_path="gs://dwh-mini-metrika/datagather/other/$(date +"%Y-%m-%d/%H")/"
      gsutil -m mv ../dwh_gs_export/*.gz "$gs_file_path"
  #    gsutil setmeta -h "Content-Type:gzip" "$gs_file_path"hits_batch*.gz
  fi
fi

# если стейдж, то удалим все данные, подготовленные для отправки в DWH
if [ "$ENVIRONMENT" = "stage" ]; then
  # узнаем, появились ли необработанные данные на стейдже
  export_bad_count=$(( export_bad_count - $(find ../dwh_gs_export -name "*.gz" | wc -l) ))
  rm -f ../dwh_gs_export/* && echo "Files from dwh_gs_export directory deleted"
fi

# если появились необработанные архивы, то отправим сообщение в google chat
if (( export_bad_count > 0 )); then
  message_first_part="{\"text\" : \"Enviroment: $ENVIRONMENT"
  message_second_part="New unprocessed data in gs://$GCP_BUCKET_SESSIONS/export_bad$hits_error$users_params_error$other_errors\"}"
  message="$message_first_part\n$message_second_part"
  curl -H 'Content-Type: application/json; charset=UTF-8' -d "$message" "https://chat.googleapis.com/v1/spaces/AAAAJGKQicQ/messages?key=AIzaSyDdI0hCZtE6vySjMm-WEfRq3CPzqKqqsHI&token=NmjRY60OxA9Uyrc7I2clV9srsMlXZuU1CxXCP9Pm-Ac%3D"
fi