Kafka日志删除源码分析

清疚 2023-01-16 15:25 178阅读 0赞

昨天[Kafka][]集群磁盘容量达到了90%，于是赶紧将Log的保存时间设置成24小时，但是发现设置完之后Log仍然没有被删除。于是今天特意去看了一下[Kafka][]日志删除相关的代码，于是有了这篇文章。

在使用[Kafka][]的时候我们一般都会根据需求对Log进行保存，比如保存1天、3天或者7天之类的，我们可以通过以下的几个参数实现：

<table> 
 <tbody> 
 <tr> 
 <td> <code>log.retention.hours</code> <code>log.retention.minutes</code> <code>log.retention.ms</code> </td> 
 </tr> 
 </tbody> 
</table>

这三个参数设置一个即可，Log的默认保存时间是168小时，也就是7天。对应的Kafka代码是这么获取这个值的：

<table> 
 <tbody> 
 <tr> 
 <td> <code>private</code> <code>def</code> <code>getLogRetentionTimeMillis()</code><code>:</code> <code>Long </code><code>=</code> <code>{ </code> <code>&nbsp;&nbsp;</code><code>val</code> <code>millisInMinute </code><code>=</code> <code>60</code><code>L * </code><code>1000</code><code>L</code> <code>&nbsp;&nbsp;</code><code>val</code> <code>millisInHour </code><code>=</code> <code>60</code><code>L * millisInMinute</code> <code>&nbsp;&nbsp;</code>&nbsp; <code>&nbsp;&nbsp;</code><code>if</code><code>(props.containsKey(</code><code>"log.retention.ms"</code><code>)){ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>props.getIntInRange(</code><code>"log.retention.ms"</code><code>, (</code><code>1</code><code>, Int.MaxValue))</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>else</code> <code>if</code><code>(props.containsKey(</code><code>"log.retention.minutes"</code><code>)){ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>millisInMinute * props.getIntInRange(</code><code>"log.retention.minutes"</code><code>, (</code><code>1</code><code>, Int.MaxValue))</code> <code>&nbsp;&nbsp;</code><code>} </code> <code>&nbsp;&nbsp;</code><code>else</code> <code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>millisInHour * props.getIntInRange(</code><code>"log.retention.hours"</code><code>, </code><code>24</code><code>*</code><code>7</code><code>, (</code><code>1</code><code>, Int.MaxValue))</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

除了配置Log保存时间，我们还可以配置Log保存的大小，可以通过`log.retention.bytes`参数设置，默认是不限制Log存储大小。

存在两种删除日志的策略，那么问题来了：删除日志的时候是两个参数都需要配置才起作用还是只需要配置其中一个？答案是只需要配置其中一个，如下代码：

<table> 
 <tbody> 
 <tr> 
 <td> <code>def</code> <code>cleanupLogs() { </code> <code>&nbsp;&nbsp;</code><code>debug(</code><code>"Beginning log cleanup..."</code><code>)</code> <code>&nbsp;&nbsp;</code><code>var</code> <code>total </code><code>=</code> <code>0</code> <code>&nbsp;&nbsp;</code><code>val</code> <code>startMs </code><code>=</code> <code>time.milliseconds</code> <code>&nbsp;&nbsp;</code><code>for</code><code>(log &lt;- allLogs; </code><code>if</code> <code>!log.config.compact) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>debug(</code><code>"Garbage collecting '"</code> <code>+ log.name + </code><code>"'"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>total +</code><code>=</code> <code>cleanupExpiredSegments(log) + cleanupSegmentsToMaintainSize(log)</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>debug(</code><code>"Log cleanup completed. "</code> <code>+ total + </code><code>" files deleted in "</code> <code>+</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>(time.milliseconds - startMs) / </code><code>1000</code> <code>+ </code><code>" seconds"</code><code>)</code> <code>}</code> &nbsp; &nbsp; <code>private</code> <code>def</code> <code>cleanupExpiredSegments(log</code><code>:</code> <code>Log)</code><code>:</code> <code>Int </code><code>=</code> <code>{ </code> <code>&nbsp;&nbsp;</code><code>val</code> <code>startMs </code><code>=</code> <code>time.milliseconds</code> <code>&nbsp;&nbsp;</code><code>log.deleteOldSegments(startMs - </code><code>_</code><code>.lastModified &gt; log.config.retentionMs)</code> <code>}</code> &nbsp; <code>private</code> <code>def</code> <code>cleanupSegmentsToMaintainSize(log</code><code>:</code> <code>Log)</code><code>:</code> <code>Int </code><code>=</code> <code>{ </code> <code>&nbsp;&nbsp;</code><code>if</code><code>(log.config.retentionSize &lt; </code><code>0</code> <code>|| log.size &lt; log.config.retentionSize)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code> <code>0</code> <code>&nbsp;&nbsp;</code><code>var</code> <code>diff </code><code>=</code> <code>log.size - log.config.retentionSize</code> <code>&nbsp;&nbsp;</code><code>def</code> <code>shouldDelete(segment</code><code>:</code> <code>LogSegment) </code><code>=</code> <code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code><code>(diff - segment.size &gt;</code><code>=</code> <code>0</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>diff -</code><code>=</code> <code>segment.size</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>true</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>} </code><code>else</code> <code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>false</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>log.deleteOldSegments(shouldDelete)</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

`cleanupLogs`函数就是清理需要删除的日志。其中调用了`cleanupExpiredSegments`和`cleanupSegmentsToMaintainSize`函数，分别对应于上面按照保存时间和保存的Log大小策略删除的，从这个可以看出，我们只需要配置一种删除策略即可。

`deleteOldSegments`函数就是根据相关的条件找出需要删除的Segments，如下：

<table> 
 <tbody> 
 <tr> 
 <td> <code>def</code> <code>deleteOldSegments(predicate</code><code>:</code> <code>LogSegment </code><code>=</code><code>&gt; Boolean)</code><code>:</code> <code>Int </code><code>=</code> <code>{ </code> <code>&nbsp;&nbsp;</code><code>// find any segments that match the user-supplied predicate </code> <code>&nbsp;&nbsp;</code><code>//UNLESS it is the final segment </code> <code>&nbsp;&nbsp;</code><code>// and it is empty (since we would just end up re-creating it</code> <code>&nbsp;&nbsp;</code><code>val</code> <code>lastSegment </code><code>=</code> <code>activeSegment</code> <code>&nbsp;&nbsp;</code><code>val</code> <code>deletable </code><code>=</code> <code>logSegments.takeWhile(s </code><code>=</code><code>&gt; predicate(s) &amp;&amp; </code> <code>&nbsp;&nbsp;</code><code>(s.baseOffset !</code><code>=</code> <code>lastSegment.baseOffset || s.size &gt; </code><code>0</code><code>))</code> <code>&nbsp;&nbsp;</code><code>val</code> <code>numToDelete </code><code>=</code> <code>deletable.size</code> <code>&nbsp;&nbsp;</code><code>if</code><code>(numToDelete &gt; </code><code>0</code><code>) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>lock synchronized { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// we must always have at least one segment, so if we are </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// going to delete all the segments, create a new one first</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>if</code><code>(segments.size </code><code>==</code> <code>numToDelete)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>roll()</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>// remove the segments for lookups</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>deletable.foreach(deleteSegment(</code><code>_</code><code>))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>numToDelete</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

`deleteOldSegments`根据传进来的predicate找出需要删除的Segments，并存放到deletable中。最后遍历deletable中的Segment，并调用`deleteSegment`函数去删除：

<table> 
 <tbody> 
 <tr> 
 <td> <code>private</code> <code>def</code> <code>deleteSegment(segment</code><code>:</code> <code>LogSegment) { </code> <code>&nbsp;&nbsp;</code><code>info(</code><code>"Scheduling log segment %d for log %s for deletion."</code> <code>.format(segment.baseOffset, name))</code> <code>&nbsp;&nbsp;</code><code>lock synchronized { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>segments.remove(segment.baseOffset)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>asyncDeleteSegment(segment)</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

而`deleteSegment`最终调用的是`asyncDeleteSegment`函数：

<table> 
 <tbody> 
 <tr> 
 <td> <code>private</code> <code>def</code> <code>asyncDeleteSegment(segment</code><code>:</code> <code>LogSegment) { </code> <code>&nbsp;&nbsp;</code><code>segment.changeFileSuffixes(</code><code>""</code><code>, Log.DeletedFileSuffix)</code> <code>&nbsp;&nbsp;</code><code>def</code> <code>deleteSeg() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>info(</code><code>"Deleting segment %d from log %s."</code><code>.format(segment.baseOffset, name))</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>segment.delete()</code> <code>&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;</code><code>scheduler.schedule(</code><code>"delete-file"</code><code>, deleteSeg, delay </code><code>=</code> <code>config.fileDeleteDelayMs)</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

从名字就可以看出，这个删除是异步进行的。从实现来看，删除是由另外一个线程执行的。删除之前会将需要删除的Log名字加上.deleted后缀：

<table> 
 <tbody> 
 <tr> 
 <td> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>8192</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000000000000</code><code>.index.deleted</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073398842</code> <code>Mar </code><code>24</code> <code>11</code><code>:</code><code>31</code> <code>00000000000000000000</code><code>.log.deleted</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>8200</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000000736086</code><code>.index.deleted</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073666711</code> <code>Mar </code><code>24</code> <code>11</code><code>:</code><code>33</code> <code>00000000000000736086</code><code>.log.deleted</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>8192</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000001468035</code><code>.index.deleted</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073555989</code> <code>Mar </code><code>24</code> <code>11</code><code>:</code><code>34</code> <code>00000000000001468035</code><code>.log.deleted</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>8192</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000002190046</code><code>.index.deleted</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073446946</code> <code>Mar </code><code>24</code> <code>11</code><code>:</code><code>36</code> <code>00000000000002190046</code><code>.log.deleted</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>26200</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000002942311</code><code>.index</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073578680</code> <code>Mar </code><code>24</code> <code>14</code><code>:</code><code>24</code> <code>00000000000002942311</code><code>.log</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>27640</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000003674166</code><code>.index</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073570675</code> <code>Mar </code><code>24</code> <code>17</code><code>:</code><code>26</code> <code>00000000000003674166</code><code>.log</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>26856</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000004398538</code><code>.index</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073623235</code> <code>Mar </code><code>24</code> <code>20</code><code>:</code><code>45</code> <code>00000000000004398538</code><code>.log</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>32480</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000005134010</code><code>.index</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073720575</code> <code>Mar </code><code>25</code> <code>01</code><code>:</code><code>36</code> <code>00000000000005134010</code><code>.log</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </code><code>54080</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>00</code> <code>00000000000005862751</code><code>.index</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog </code><code>1073535666</code> <code>Mar </code><code>25</code> <code>10</code><code>:</code><code>10</code> <code>00000000000005862751</code><code>.log</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp;&nbsp; </code><code>10485760</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>36</code> <code>00000000000006599973</code><code>.index</code> <code>-rw-r--r-- </code><code>1</code> <code>iteblog iteblog&nbsp; </code><code>545669095</code> <code>Mar </code><code>25</code> <code>11</code><code>:</code><code>37</code> <code>00000000000006599973</code><code>.log</code> </td> 
 </tr> 
 </tbody> 
</table>

然后会经过`log.segment.delete.delay.ms`(默认1分钟)时间之后彻底删除那些Segments。

Kafka集群会每隔`log.retention.check.interval.ms`(默认5分钟)时间去检测需要删除的Segments。

<table> 
 <tbody> 
 <tr> 
 <td> <code>scheduler.schedule(</code><code>"kafka-log-retention"</code><code>, </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>cleanupLogs, </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>delay </code><code>=</code> <code>InitialTaskDelayMs, </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>period </code><code>=</code> <code>retentionCheckMs, </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>TimeUnit.MILLISECONDS)</code> </td> 
 </tr> 
 </tbody> 
</table>

[Kafka]: https://www.iteblog.com/archives/tag/kafka/