はじめに

GKEで Fluentd について調べていたところ、fluentd-gcp-scaler というリソースがありました。

$ kubectl get po -n kube-system | grep fluentd
fluentd-gcp-scaler-8b674f786-k8989                 1/1       Running   0          2m
fluentd-gcp-v3.2.0-2rszh                           2/2       Running   0          2m
fluentd-gcp-v3.2.0-klv6r                           2/2       Running   0          2m
fluentd-gcp-v3.2.0-z65k7                           2/2       Running   0          2m

気になって GitHub のソースコードを読んでみたので、内容をまとめます。

構成

fluentd-gcp-scaler は以下の4つのファイルで構成されています。

DockerFile
Makefile
example-deployment.yaml
scaler.sh

DockerFile

fluentd-gcp-scaler のイメージ用 DockerFile。
コンテナ内で次の処理を行います。

curl のインストール
curl を使用した kubectl のインストール
curl の削除
scaler.sh の配置

Makefile

Docker イメージのビルド＆プッシュを行います。
記載時点のバージョンは0.5.1

example-deployment.yaml

fluentd-gcp-scaler のデプロイ用マニフェストファイルのサンプル。
デプロイしたコンテナに対して、以下の処理を行います。

環境変数の設定

$ CPU_REQUEST=100m
$ MEMORY_REQUEST=200Mi
$ MEMORY_LIMIT=300Mi

スクリプトの実行

scaler.sh をオプション付きで実行。

$ /scaler.sh --ds-name fluentd-gcp-v2.0.13 --scaling-policy fluentd-gcp-scaling-policy

scaler.sh

fluentd-gcp-scaling-policy の本体となるshellスクリプト。
機能ごとにコードを見ていきます。

変数の定義

SLEEP_SECONDS=${SLEEP_SECONDS:-60}
NAMESPACE=${NAMESPACE:-kube-system}
DEFAULT_CPU_REQUEST=${CPU_REQUEST:-}
DEFAULT_MEMORY_REQUEST=${MEMORY_REQUEST:-}
DEFAULT_CPU_LIMIT=${CPU_LIMIT:-}
DEFAULT_MEMORY_LIMIT=${MEMORY_LIMIT:-}

パラメータのチェック

while test $# -gt 0; do
  case "$1" in
    --ds-name=*)
      export DS_NAME=$(echo $1 | sed -e 's/^[^=]*=//g')
      if [ -z ${DS_NAME} ]; then
        log "Missing DaemonSet name in --ds-name flag." >&2
        exit 1
      fi
      shift
      ;;
    --scaling-policy=*)
      export SCALING_POLICY=$(echo $1 | sed -e 's/^[^=]*=//g')
      if [ -z ${SCALING_POLICY} ]; then
        log "Missing ScalingPolicy name in --scaling-policy flag." >&2
        exit 1
      fi
      shift
      ;;
    *)
      log "Unrecognized argument $1." >&2
      exit 1
      ;;
  esac
done

testコマンドでパラメータ数をカウントし、パラメータが存在する場合は次の操作を行います。

オプションがds-nameの時、環境変数DS_NAMEに設定
設定したDS_NAMEが空の場合はエラーを出力し終了
オプションがscaling-policyの場合、環境変数SCALING_POLICYに設定
設定したSCALING_POLICYが空の場合はエラーを出力し終了
それ以外の場合はエラーを出力し終了

ロジックとしては「第1パラメータを判定 → shiftコマンドで次のパラメータにシフト」を繰り返します。

if [ -z ${DS_NAME} ]; then
  log "DaemonSet name has to be set via --ds-name flag." >&2
  exit 1
fi

if [ -z ${SCALING_POLICY} ]; then
  log "ScalingPolicy name has to be set via --scaling-policy flag." >&2
  exit 1
fi

環境変数DS_NAME、SCALING_POLICYが空の場合はエラーを出力し処理を終了します。
オプションにds-name、scaling-policyのいずれかが存在しない場合はエラーとなります。

メイン処理

while true
do
  reset_to_defaults
  apply_scaling
  build_flags
  update_if_needed
  sleep ${SLEEP_SECONDS}
done

SLEEP_SECONDSのインターバルでループ処理を実行します。（デフォルトは60秒）
ループ内では4つの関数を順に呼び出しています。

reset_to_defaults()

reset_to_defaults() {
  REQUESTS_FLAG=
  LIMITS_FLAG=
  CPU_REQUEST=${DEFAULT_CPU_REQUEST}
  MEMORY_REQUEST=${DEFAULT_MEMORY_REQUEST}
  CPU_LIMIT=${DEFAULT_CPU_LIMIT}
  MEMORY_LIMIT=${DEFAULT_MEMORY_LIMIT}
}

フラグ値を空にし、変数にデフォルト値を設定。

apply_scaling()

apply_scaling() {
  # This is assuming there is a ScalingPolicy installed in the cluster.
  # See https://github.com/justinsb/scaler for more details.
  if ! kubectl get scalingpolicies -n ${NAMESPACE} ${SCALING_POLICY} 2> /dev/null
  then
    return
  fi
  for resource_class in request limit
  do
    for resource_type in cpu memory
    do
      TMP=$(kubectl get scalingpolicies -n ${NAMESPACE} ${SCALING_POLICY} \
        -o=jsonpath="{.spec.containers[?(@.name=='fluentd-gcp')].resources.${resource_class}s[?(@.resource=='${resource_type}')].base}")
      if [ ${TMP} ]; then
        # Build the right variable name from resource_type and resource_class
        # and assign it.
        export $(echo ${resource_type}_${resource_class} | awk '{print toupper($0)}')=${TMP}
      fi
    done
  done
}

ScalingPolicy リソースから取得した値を、以下の変数に設定します。

CPU_REQUEST
CPU_LIMIT
MEMORY_REQUEST
MEMORY_LIMIT

resource_class(request, limit)とresource_type(cpu, memory)の二重ループ処理で変数名を生成しています。
ScalingPolicy が存在しない場合は何も行いません。

ScalingPolicy(イメージ)

# ScalingPolicy にはCPU、メモリの条件が設定されている
spec:
  containers:
    - name: fluentd-gcp
      resources:
        - requests:
          - resource: cpu
            base: 500m
          - resource: memory
            base: 512Mi
        - limits:
          - resource: cpu
            base: 1000m
          - resource: memory
            base: 1024Mi

build_flags()

build_flags() {
  add_request cpu ${CPU_REQUEST}
  add_request memory ${MEMORY_REQUEST}
  add_limit cpu ${CPU_LIMIT}
  add_limit memory ${MEMORY_LIMIT}
}

apply_scaling()で設定した環境変数を使用し、以下の変数に値を設定します。

REQUESTS_FLAG
LIMITS_FLAG

関数add_request()とadd_limit()はそれぞれの変数に値を追加していきます。

REQUESTS_FLAG=--requests=cup=500m,memory=512Mi
LIMITS_FLAG=--limits=cup=1000m,memory=1024Mi

update_if_needed()

update_if_needed() {
  NEED_UPDATE=false
  if needs_update requests.cpu ${CPU_REQUEST}; then NEED_UPDATE=true; fi
  if needs_update requests.memory ${MEMORY_REQUEST}; then NEED_UPDATE=true; fi
  if needs_update limits.cpu ${CPU_LIMIT}; then NEED_UPDATE=true; fi
  if needs_update limits.memory ${MEMORY_LIMIT}; then NEED_UPDATE=true; fi
  if ! ${NEED_UPDATE}
  then
    return
  fi
  if [ ${REQUESTS_FLAG} ] || [ ${LIMITS_FLAG} ]
  then
    KUBECTL_CMD="kubectl set resources -n ${NAMESPACE} ds ${DS_NAME} -c fluentd-gcp ${REQUESTS_FLAG} ${LIMITS_FLAG}"
    log "Running: ${KUBECTL_CMD}"
    ${KUBECTL_CMD}
  fi
}

関数needs_update()で

DaemonSet の現在のCPU、メモリ設定値
ScalingPolicy によるCPU、メモリ要求値

を比較し、差異がある場合にアップデートを行います。
アップデートにはkubectlコマンドを使用します。

# コマンド例
kubectl set resources -n kube-system ds fluentd-gcp-v2.0.13 -c fluentd-gcp \
--requests=cup=500m,memory=512Mi --limits=cup=1000m,memory=1024Mi

コマンドリファレンス

コマンド	意味
[ 文字列 ]	文字列の長さが0より大きければTrue
[ -z 文字列 ]	文字列の長さが0であればTrue
$#	パラメータ数
$0	実行プログラム
\$1 ~ $9	プログラムに渡したパラメータ
>&2	標準エラー出力
2> /dev/null	標準エラー出力を破棄
shift	先頭のパラメータを破棄し、以降を先頭にシフト

JSONPath Support - Kubernetes

まとめ

ScalingPolicy に値を設定することで、Fluentd の使用リソースを変更する機能のようです。
中身がわかりスッキリしました。

shellスクリプトには不慣れだったので、良い勉強になりました。

fluentd-gcp-scaler コードリーディング