connector-restart

#!/usr/bin/env bash

set -euo pipefail

# Logging level
: "${LOG_LEVEL:=INFO}"

# Associative arrays for logging
declare -A log_c log_l
# Color codes
log_c[0]='\033[0m'          # Reset colors
log_c[error]='\033[1;31m'   # Red
log_c[warning]='\033[1;33m' # Yellow
log_c[info]='\033[1;34m'    # Blue
log_c[success]='\033[1;32m' # Green
log_c[debug]='\033[1;37m'   # Gray
log_c[none]=''              # Nothing
# Logging priorities
log_l[silent]=0 # Higest prority, suppress all
log_l[error]=1
log_l[warning]=2
log_l[info]=3
log_l[success]=3
log_l[debug]=4 # Lowest prority, print all

# Print UTC date with format like: [05/17/21 20:05:21 UTC]
date-fmt() { date '+%D %T %Z' -u; }

log() {
  local lvl="${1:-debug}" color="${1:-debug}" msg="${*:2}" accent=debug ident=7

  # Check log level argument is valid, or set to debug
  [[ " ${!log_l[*]} " =~ \ "${lvl,,}"\  ]] || lvl=debug

  # Check interactive tty
  ! [ -t 1 ] && color=none && accent=none && ident=1

  # Redirect all messages to the third stream
  exec 3>&1
  # Redirect errors to stderr
  [ "${lvl,,}" == error ] && exec 3>&2
  # If the log level is >= to the msg level then not suppress messages
  [ "${log_l[${LOG_LEVEL,,}]}" -ge "${log_l[${lvl,,}]}" ] || exec 3>/dev/null

  # Print log message
  printf >&3 \
    "${log_c[${color,,}]}%-${ident}s${log_c[$accent]} [%s] ${log_c[0]}%s\n" \
    "${lvl^^}" "$(date-fmt)" "$msg"

  # Exiting the script when an error or success message appears
  if [ "${lvl,,}" == error ]; then
    exit 1
  fi

  if [ "${lvl,,}" == success ]; then
    exit 0
  fi
}

get_connectors() {
  curl -Lff "$_connect_user_pass" "$_connect_url/connectors?expand=status"
}

parse_failed_connectors() {
  jq -rc -M '
    map({name: .status.name} + {connector: .status.connector})
    | .[]| {connect: ((.connector) + {name: .name})}
    | select(.connect.state=="FAILED")
    | ("/connectors/"+ .connect.name + "/restart")
  '
}

parse_failed_tasks() {
  jq -rc -M '
    map({name: .status.name } + {tasks: .status.tasks} + {connector: .status.connector})
    | .[] | {task: ((.tasks[]) + {name: .name}), connector: .connector}
    | select(.task.state=="FAILED" and .connector.state != "PAUSED") |
    {name: .task.name, task_id: .task.id|tostring}
    | ("/Connectors/"+ .name + "/tasks/" + .task_id + "/restart")
  '
}

state_connectors_and_tasks() {
  jq -r -M '
    map({name: .status.name} + {connector: .status.connector} + {tasks: .status.tasks})
    | .[]| {connect: ((.connector) + {name: .name})} + {task: ((.tasks[]) + {name: .name})}
    | ("Connector "+ .connect.name + " state is " + .connect.state + " and tasks state " + .task.state)
  '
}

#Set vars
: "${SIDECAR_MODE:=false}"
: "${KAFKA_CONNECT_HOST:=localhost}"
: "${KAFKA_CONNECT_PROTO:=http}"
: "${KAFKA_CONNECT_PORT:=8083}"
: "${KAFKA_CONNECT_USER:=}"
: "${KAFKA_CONNECT_PASS:=}"
: "${REQUEST_DELAY:=30}"
_connect_url="$KAFKA_CONNECT_PROTO://$KAFKA_CONNECT_HOST:$KAFKA_CONNECT_PORT"
: "${_connect_user_pass:=}"

if [ -z "$KAFKA_CONNECT_USER" ] && [ -z "$KAFKA_CONNECT_PASS" ]; then
  _connect_user_pass="-u $KAFKA_CONNECT_USER:$KAFKA_CONNECT_PASS"
fi

#Print vars
log info SIDECAR_MODE="$SIDECAR_MODE"
log info KAFKA_CONNECT_PROTO="$KAFKA_CONNECT_PROTO"
log info KAFKA_CONNECT_HOST="$KAFKA_CONNECT_HOST"
log info KAFKA_CONNECT_PORT="$KAFKA_CONNECT_PORT"
log info REQUEST_DELAY="$REQUEST_DELAY"

if [ "$SIDECAR_MODE" = true ]; then
  #wait start kafka-connect
  log info 'Wait start kafka-connect 60 second'
  sleep 60s

  while true; do
    log info 'Get status connectors'

    if connectorsjson=$(get_connectors); then
      log info 'Status connectors already'

      # List current connectors and status
      mapfile -t aray_current_states < <(
        state_connectors_and_tasks <<<"$connectorsjson"
      )

      log info "${aray_current_states[*]}"

      # Restart any connectors that are FAILED
      parse_failed_connectors <<<"$connectorsjson" |
        xargs -I"{connectors}" \
          curl -v -X POST "$_connect_user_pass" "$_connect_url"\{connectors\}

      sleep 10s
      connectorsjson=$(get_connectors)

      # Restart any connector tasks that are FAILED
      parse_failed_tasks <<<"$connectorsjson" |
        xargs -I"{connector_and_task}" \
          curl -v -X POST "$_connect_user_pass" \
          "$_connect_url"\{connector_and_task\}

      # log restart connectors
      mapfile -r aray_failed_tasks < <(
        jq -rc -M '
          map({name: .status.name } + {tasks: .status.tasks} + {connector: .status.connector})
          | .[] | {task: ((.tasks[]) + {name: .name}), connector: .connector}
          | select(.task.state=="FAILED" and .connector.state != "PAUSED")
          | {name: .task.name, task_id: .task.id | tostring}
          | ("Connector: " + .name + " | Tasks: " + .task_id + " - RESTART")
        ' <<<"$connectorsjson"
      )

      log info "${aray_failed_tasks[*]}"

    else
      log error \
        "Connect to API Kafka Connect $_connect_url failed" \
        "with status code $? and message $connectorsjson"
    fi

    log info "Wait next check after $REQUEST_DELAY minut(s)"
    sleep "$REQUEST_DELAY"m
  done

else

  if connectorsjson=$(get_connectors); then
      log info 'Status connectors already'
      # List current connectors and status
      mapfile -t aray_current_states < <(
        state_connectors_and_tasks <<<"$connectorsjson"
      )

      log info "${aray_current_states[*]}"

    # Restart any connectors that are FAILED and not PAUSED
    parse_failed_connectors <<<"$connectorsjson" |
      xargs -I"{connectors}" \
        curl -v -X POST "$_connect_user_pass" "$_connect_url"\{connectors\}

    sleep 10s
    connectorsjson=$(get_connectors)

    # Restart any connector tasks that are FAILED
    parse_failed_tasks <<<"$connectorsjson" |
      xargs -I"{connector_and_task}" \
        curl -v -X POST "$_connect_user_pass" "$_connect_url"\{connector_and_task\}

  else
    log error \
      "Connect to API Kafka Connect $_connect_url failed" \
      "with status code $? and message $connectorsjson"
  fi
fi