PV사용량 모니터링하기

안녕하세요. yeTi입니다.
오늘은 Kubernetes의 PV 사용량을 모니터링하려고 합니다.

Kubernetes Storage Management Layer

대략적으로 kubernetes의 storage를 관리하는 구조가 어떻게 이뤄져 있는지에 대해 살펴보겠습니다.

위의 그림과 같이 PVC에 개발자가 사용하고 싶은 Storage의 스펙을 정의합니다.
그러면 Storage 관리자가 PV를 정의하여 물리 disk를 할당하고 Storage Class에 저장소의 특성을 정의합니다.
그러면 Provisioner에 의해서 다양한 Storage에 접근할 수 있고 최종적으로 실제 데이터는 Storage에 저장됩니다.

여기서 Storage로 ceph을 사용하고 있기 때문에 Provisioner에 rook-ceph을 넣어놨습니다.

CLI 환경

CLI 환경에서 현황을 확인해보겠습니다.

PV 설정 용량은 다음과 같이 확인할 수 있습니다.

$ kubectl describe pvc -n [namespace] [pvc name] | grep Capacity
Capacity:      5Gi
$ kubectl describe pv [pv name] | grep Capacity
Capacity:      5Gi

PV 사용량은 다음과 같이 확인할 수 있습니다.

$ sudo du -sh /var/lib/kubelet/pods/[Pod UID]/volumes/kubernetes.io~csi/[PV Name]/mount/logs

Container의 Volume 사용량은 다음과 같이 서비스 특정에 맞게 확인할 수 있습니다. 본 예제에서는 kafka를 대상으로 진행했습니다.

$ kubectl exec [Pod Name] --container kafka-broker -n [Namespace] -- du -sh /opt/kafka/data/logs

Prometheus 활용

CLI환경에서도 PV의 상태를 확인할 수 있지만 Prometheus를 활용하여 GUI환경에서 모니터링하는 환경을 구축할 수 있습니다.

설치되어 있는 Grafana의 Explore 메뉴에 접근하면 다음과 같이 PromQL로 조회할 수 있는 환경을 제공합니다.

다음 지표들을 활용해 모니터링 환경을 구축할 수 있습니다.

• kube_pod_spec_volumes_persistentvolumeclaims_info : Pod이 가지고 있는 PVC 정보 제공
• kube_persistentvolumeclaim_info : PVC와 PV 맵핑 정보
• kube_persistentvolume_capacity_bytes : PV의 설정 용량
• kafka_log_log_size : Kafka의 토픽별 로그 용량

PromQL

간략하게 Prometheus에서 사용할 수 있는 PromQL의 예제를 공유하겠습니다.

컬럼 추가

kube_persistentvolumeclaim_info에서 volumename컬럼의 데이터들을 persistentvolume 컬럼으로 추가해라

label_replace(kube_persistentvolumeclaim_info{namespace="kafka"}, "persistentvolume", "$0", "volumename", ".+")

Inner Join

kube_persistentvolumeclaim_info와 kube_persistentvolume_capacity_bytes을 persistentvolume로 Inner Join해서 kube_persistentvolume_capacity_bytes의 value를 사용해라

label_replace(kube_persistentvolumeclaim_info{namespace="kafka"}, "persistentvolume", "$0", "volumename", ".+") * on (persistentvolume) kube_persistentvolume_capacity_bytes

해당 쿼리에서 kube_persistentvolumeclaim_info의 컬럼을 사용하고 싶다면 group_left() 추가

label_replace(kube_persistentvolumeclaim_info{namespace="kafka"}, "persistentvolume", "$0", "volumename", ".+") * on (persistentvolume) group_left() kube_persistentvolume_capacity_bytes

다른 두 metric의 연산

/ on (pod)을 기준으로 앞/뒤 메트릭을 연산한다. /연산을 수행하고 pod컬럼을 기준으로 연산한다.

((sum(kafka_log_log_size{pod=~"kafka-."}) by (pod) / 1024 / 1024 / 1024) + 1.3) / on (pod) (kube_pod_spec_volumes_persistentvolumeclaims_info{namespace="kafka"} * on (persistentvolumeclaim) group_left() (label_replace(kube_persistentvolumeclaim_info{namespace="kafka"}, "persistentvolume", "$0", "volumename", ".+") * on (persistentvolume) group_left() kube_persistentvolume_capacity_bytes) / 1024 / 1024 / 1024) * 100

Prometheus의 지표들과 PromQL을 활용하여 최종적으로 다음과 같은 GUI 환경을 구성할 수 있습니다.

Troubleshooting

java.io.IOException: Disk quota exceeded

Kafka에서 데이터를 처리할때 Disk가 꽉차서 발생하는 오류입니다.

Kafka를 활용함에 있어서 다음과 같이 두가지 경우의 오류를 확인할 수 있었습니다.

ERROR Error while appending records to __consumer_offsets-9 in dir /opt/kafka/data/logs (kafka.server.LogDirFailureChannel)
java.io.IOException: Disk quota exceeded
    at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
    at sun.nio.ch.FileDispatcherImpl.write(FileDispatcherImpl.java:60)
    at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
    at sun.nio.ch.IOUtil.write(IOUtil.java:65)
    at sun.nio.ch.FileChannelImpl.write(FileChannelImpl.java:211)
    at org.apache.kafka.common.record.MemoryRecords.writeFullyTo(MemoryRecords.java:90)
    at org.apache.kafka.common.record.FileRecords.append(FileRecords.java:164)
    at kafka.log.LogSegment.append(LogSegment.scala:138)
    at kafka.log.Log$$anonfun$append$2.apply(Log.scala:980)
    at kafka.log.Log$$anonfun$append$2.apply(Log.scala:857)
    at kafka.log.Log.maybeHandleIOException(Log.scala:2058)
    at kafka.log.Log.append(Log.scala:857)
    at kafka.log.Log.appendAsFollower(Log.scala:837)
    at kafka.cluster.Partition$$anonfun$doAppendRecordsToFollowerOrFutureReplica$1.apply(Partition.scala:708)
    at kafka.cluster.Partition$$anonfun$doAppendRecordsToFollowerOrFutureReplica$1.apply(Partition.scala:699)
    at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251)
    at kafka.utils.CoreUtils$.inReadLock(CoreUtils.scala:257)
    at kafka.cluster.Partition.doAppendRecordsToFollowerOrFutureReplica(Partition.scala:698)
    at kafka.cluster.Partition.appendRecordsToFollowerOrFutureReplica(Partition.scala:715)
    at kafka.server.ReplicaFetcherThread.processPartitionData(ReplicaFetcherThread.scala:157)
    at kafka.server.AbstractFetcherThread$$anonfun$kafka$server$AbstractFetcherThread$$processFetchRequest$2$$anonfun$apply$mcV$sp$3$$anonfun$apply$9.apply(AbstractFetcherThread.scala:310)
    at kafka.server.AbstractFetcherThread$$anonfun$kafka$server$AbstractFetcherThread$$processFetchRequest$2$$anonfun$apply$mcV$sp$3$$anonfun$apply$9.apply(AbstractFetcherThread.scala:300)
    at scala.Option.foreach(Option.scala:257)
    at kafka.server.AbstractFetcherThread$$anonfun$kafka$server$AbstractFetcherThread$$processFetchRequest$2$$anonfun$apply$mcV$sp$3.apply(AbstractFetcherThread.scala:300)
    at kafka.server.AbstractFetcherThread$$anonfun$kafka$server$AbstractFetcherThread$$processFetchRequest$2$$anonfun$apply$mcV$sp$3.apply(AbstractFetcherThread.scala:299)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
    at kafka.server.AbstractFetcherThread$$anonfun$kafka$server$AbstractFetcherThread$$processFetchRequest$2.apply$mcV$sp(AbstractFetcherThread.scala:299)
    at kafka.server.AbstractFetcherThread$$anonfun$kafka$server$AbstractFetcherThread$$processFetchRequest$2.apply(AbstractFetcherThread.scala:299)
    at kafka.server.AbstractFetcherThread$$anonfun$kafka$server$AbstractFetcherThread$$processFetchRequest$2.apply(AbstractFetcherThread.scala:299)
    at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251)
    at kafka.server.AbstractFetcherThread.kafka$server$AbstractFetcherThread$$processFetchRequest(AbstractFetcherThread.scala:298)
    at kafka.server.AbstractFetcherThread$$anonfun$maybeFetch$1.apply(AbstractFetcherThread.scala:132)
    at kafka.server.AbstractFetcherThread$$anonfun$maybeFetch$1.apply(AbstractFetcherThread.scala:131)
    at scala.Option.foreach(Option.scala:257)
    at kafka.server.AbstractFetcherThread.maybeFetch(AbstractFetcherThread.scala:131)
    at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:113)
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)

ERROR Error while deleting the clean shutdown file in dir /opt/kafka/data/logs (kafka.server.LogDirFailureChannel)
java.io.IOException: Disk quota exceeded
    at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
    at sun.nio.ch.FileDispatcherImpl.write(FileDispatcherImpl.java:60)
    at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
    at sun.nio.ch.IOUtil.write(IOUtil.java:65)
    at sun.nio.ch.FileChannelImpl.write(FileChannelImpl.java:211)
    at kafka.log.ProducerStateManager$.kafka$log$ProducerStateManager$$writeSnapshot(ProducerStateManager.scala:458)
    at kafka.log.ProducerStateManager.takeSnapshot(ProducerStateManager.scala:687)
    at kafka.log.Log.rebuildProducerState(Log.scala:717)
    at kafka.log.Log.kafka$log$Log$$recoverSegment(Log.scala:499)
    at kafka.log.Log.kafka$log$Log$$recoverLog(Log.scala:615)
    at kafka.log.Log$$anonfun$2.apply$mcJ$sp(Log.scala:575)
    at kafka.log.Log$$anonfun$2.apply(Log.scala:575)
    at kafka.log.Log$$anonfun$2.apply(Log.scala:575)
    at kafka.log.Log.retryOnOffsetOverflow(Log.scala:2069)
    at kafka.log.Log.loadSegments(Log.scala:574)
    at kafka.log.Log.<init>(Log.scala:292)
    at kafka.log.Log$.apply(Log.scala:2202)
    at kafka.log.LogManager.kafka$log$LogManager$$loadLog(LogManager.scala:265)
    at kafka.log.LogManager$$anonfun$loadLogs$2$$anonfun$11$$anonfun$apply$15$$anonfun$apply$2.apply$mcV$sp(LogManager.scala:345)
    at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:63)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

해결

Kafka를 직접사용하고 있다면 /opt/kafka/data/logs, Container형태로 사용하고 있다면 마운트한 볼륨위치에 가면 토픽명들을 확인할 수 있습니다.

이 파일들은 토픽의 로그들인데 적절하게 지우면 용량을 확보할 수 있습니다.