本文基于Elasticsearch7.10.2分析

0.Ingest 节点 概述

在实际进行文档index之前，使用采集节点（默认情况下，每个es节点都是ingest）对文档进行预处理。采集节点会拦截bulk和index请求，进行转换，然后将文档传回index或bulk API。

每个索引都有index.default_pipeline 和 index.final_pipeline 两个配置。

他们都是用于指定 Elasticsearch index 或者bulk 文档时要执行的预处理逻辑。

• index.default_pipeline 定义了默认管道，它会在索引文档时首先执行。但如果索引请求中指定了 pipeline 参数，则该参数指定的管道会覆盖默认管道。如果设置了 index.default_pipeline 但对应的管道不存在，索引请求会失败。特殊值 _none 表示不运行任何摄取管道。
  
• index.final_pipeline 定义了最终管道，它总是在请求管道（如果指定）和默认管道（如果存在）之后执行。如果设置了 index.final_pipeline 但对应的管道不存在，索引请求会失败。特殊值 _none 表示不运行最终管道。
  
  简而言之，default_pipeline 先执行，可被覆盖；final_pipeline 后执行，不可被覆盖。两者都可设为 _none 以禁用。
  
  一个Pipeline可以有多个Processor组成，每个Processor有着各自的不同功能，官方支持的Processor可参考：https://www.elastic.co/guide/e ... .html
  
  一个简单的例子,利用Set Processor 对新增的文档中加入新的字段和值：
  
  java<br /> PUT _ingest/pipeline/set_os<br /> {<br /> "description": "sets the value of host.os.name from the field os",<br /> "processors": [<br /> {<br /> "set": {<br /> "field": "host.os.name", // 增加的属性<br /> "value": "{{os}}" // 这里引用了文档原先的os属性， 这里可以直接填写其他值<br /> }<br /> }<br /> ]<br /> }<br /> <br /> POST _ingest/pipeline/set_os/_simulate<br /> {<br /> "docs": [<br /> {<br /> "_source": {<br /> "os": "Ubuntu"<br /> }<br /> }<br /> ]<br /> }<br />
  
  这样转换之后，文档内容就变成了：
  
  java<br /> {<br /> "host" : {<br /> "os" : {<br /> "name" : "Ubuntu"<br /> }<br /> },<br /> "os" : "Ubuntu"<br /> }<br />
  
  

  写单个文档的流程概述
  

  
  

  

  
  
  
  当请求，或者索引本身配置有pipline的时候，协调节点就会转发到ingest节点
  
  java<br /> PSOT source_index/_doc?pipeline=set_os<br /> {<br /> "os": "xxxx"<br /> }<br />
  
  

  

  
  
  
  
  【注】 并不是一定会发生内部rpc的请求转发，如果本地节点能接受当前的请求则不会转发到其他节点。
  
  

  1. 模块总体概述
  

  
  本小节关注IngestService中重要的相关类，对这些类有一个整体的了解有助于理解该模块。
  
  

• ClusterService
  
  • IngestService 实现了 ClusterStateApplier 接口， 这样就能监听和响应集群的状态变化，当集群状态更新时，IngestService可以调整其内部 pipelines完成CRUD。
    
  • 另外IngestService还有List<Consumer>用来对提供给对集群状态变更之后需要最新状态的插件。
    
• ScriptService
  
  • 某些Processor需要其用于管理和执行脚本，比如Script Processor。
    
• AnalysisRegistry
  
  • 某些需要对文档内容进行分词处理的Processor。
    
• ThreadPool
  
  • Processor都是异步执行的，实际执行线程池取决于调用上下文（如 write 或 management）
    
    • bulk API 时发生的pipeline 处理使用的是write线程池。
      
    • pipeline/_simulate API 使用的是management线程池，模拟执行通常是短时间的、低频的任务，不需要高并发支持而且为了不影响实际的文档处理或其他重要任务。
      
  • 另外为了避免Grok Processor运行时间过长，使用了Generic线程做定时调度检查执行时间
    
• IngestMetric
  
  • 通过实现ReportingService接口来做到展示ingest内部的执行情况。
    
  • GET _nodes/stats?filter_path=nodes.*.ingest 可以查看到ingest 中的每个Pipeline中的执行的次数、失败次数以及总耗时
    
• IngestPlugin
  
  • Ingest支持加载自定义的Processor插件，系统内置的所有Processor以及自定义的都通过IngestService 中的Map<String, Processor.Factory> processorFactories来进行管理。
    
• IngestDocument
  
  • 其包含文档的源数据，提供了修改和查询文档的字段的能力，为Processor灵活操作文档数据提供基础，在后续pipeline的执行中，也是由其的executePipeline方法驱动的。
    
    

    2. Processor 实现机制
    

    
    

    抽象工厂设计模式的应用
    

    
    

    Processor接口设计
    

    
    每个Processor都有核心方法execute，使得处理器能够以统一的方式操作 IngestDocument，并通过多态实现不同处理逻辑。
    
    

    Processor.Factory的设计
    

    
    Processor.Factory 是 Processor 的抽象工厂，负责动态创建处理器实例。其主要职责包括：
    
    

• 动态实例化：
  
  • Processor create() 方法接收处理器配置并创建具体的处理器。
    
  • 支持递归创建，例如ConditionalProcessor可以嵌套其他处理器。
    
• 依赖注入：
  
  • Processor.Parameters 提供了一组服务和工具（如 ScriptService），工厂可以利用这些依赖创建复杂的Processor。
    
    

    Processor.Factory的集中管理
    

    
    在 IngestService 中，通过 Map<String, Processor.Factory> processorFactories 集中管理所有处理器工厂。这种管理方式提供了以下优势：
    
    动态扩展：
    
    

• 插件可以注册自定义Processor工厂。
  
• 新处理器类型的注册仅需添加到 processorFactories。
  
  

  组合以及装饰器设计模式的应用
  

  
  Processor经典的几个类的关系如下：
  
  

  

  
  
  
  
  CompoundProcessor是经典的组合设计模式，Pipeline这个类可以像使用单个 Processor 一样调用 CompoundProcessor，无需关注其内部具体细节。
  
  而ConditionalProcessor 以及TrackingResultProcessor则体现了装饰器模式，在不改变原有对象的情况下扩展功能：
  
  

• ConditionalProcessor 在执行Processor前会调用evaluate方法判断是否需要执行。
  
• TrackingResultProcessor中decorate是为 CompoundProcessor 及其内部的 Processor 添加跟踪功能。
  
  

  如何自定义Processor 插件
  

  
  自定义 Processor 插件的注册方式为实现 IngestPlugin （Elasticsearch 提供不同的插件接口用来扩展不同类型的功能）的 getProcessors 方法，该方法返回一个工厂列表，IngestService 会将这些工厂注入到 processorFactories 中。
  
  分析完代码之后，我们回到实战中来，简单起见，我们实现类似Append的Processor，但是我们这个更简单，输入的是字符串，然后我们用,分割一下将其作为数组设为值。
  
  java<br /> import org.elasticsearch.ingest.Processor;<br /> import org.elasticsearch.plugins.IngestPlugin;<br /> import org.elasticsearch.plugins.Plugin;<br /> import java.util.HashMap;<br /> import java.util.Map;<br /> public class AddArrayProcessorPlugin extends Plugin implements IngestPlugin {<br /> @Override<br /> public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {<br /> Map<String, Processor.Factory> processors = new HashMap<>();<br /> processors.put(AddArrayProcessor.TYPE, new AddArrayProcessor.Factory());<br /> return processors;<br /> }<br /> }<br />

java<br /> import org.elasticsearch.ingest.AbstractProcessor;<br /> import org.elasticsearch.ingest.ConfigurationUtils;<br /> import org.elasticsearch.ingest.IngestDocument;<br /> import org.elasticsearch.ingest.Processor;<br /> import java.util.*;<br /> <br /> public class AddArrayProcessor extends AbstractProcessor {<br /> public static final String TYPE = "add_array";<br /> public String field;<br /> public String value;<br /> protected AddArrayProcessor(String tag, String description, String field, String value) {<br /> super(tag, description);<br /> this.field = field;<br /> this.value = value;<br /> }<br /> @Override<br /> public IngestDocument execute(IngestDocument ingestDocument) throws Exception {<br /> List valueList = new ArrayList<>(Arrays.asList(value.split(",")));<br /> ingestDocument.setFieldValue(field, valueList);<br /> return ingestDocument;<br /> }<br /> @Override<br /> public String getType() {<br /> return TYPE;<br /> }<br /> public static final class Factory implements Processor.Factory {<br /> @Override<br /> public Processor create(Map<String, Processor.Factory> processorFactories, String tag,<br /> String description, Map<String, Object> config) throws Exception {<br /> String field = ConfigurationUtils.readStringProperty(TYPE, tag, config, "field");<br /> String value = ConfigurationUtils.readStringProperty(TYPE, tag, config, "value");<br /> return new AddArrayProcessor(tag, description, field, value);<br /> }<br /> }<br /> }<br />
  
  打包之后，我们去install我们的Processor插件：
  
  java<br /> bin/elasticsearch-plugin install file:///home/hcb/data/data/es/data_standalone/elasticsearch-7.10.2-SNAPSHOT/AddArrayProcessor-1.0-SNAPSHOT.zip <br /> warning: no-jdk distributions that do not bundle a JDK are deprecated and will be removed in a future release<br /> -> Installing file:///home/hcb/data/data/es/data_standalone/elasticsearch-7.10.2-SNAPSHOT/AddArrayProcessor-1.0-SNAPSHOT.zip<br /> -> Downloading file:///home/hcb/data/data/es/data_standalone/elasticsearch-7.10.2-SNAPSHOT/AddArrayProcessor-1.0-SNAPSHOT.zip<br /> [=================================================] 100%   <br /> -> Installed AddArrayProcess<br />
  
  测试一下：
  
  java<br /> POST /_ingest/pipeline/_simulate<br /> {<br /> "pipeline": {<br /> "processors": [<br /> {<br /> "add_array": {<br /> "field": "test_arr",<br /> "value": "a,b,c,d,e,f"<br /> }<br /> }<br /> ]<br /> },<br /> "docs": [<br /> {<br /> "_index": "test",<br /> "_id": "1",<br /> "_source": {<br /> "field": "value"<br /> }<br /> }<br /> ]<br /> }<br /> docs结果：<br /> "_source" : {<br /> "field" : "value",<br /> "test_arr" : [<br /> "a",<br /> "b",<br /> "c",<br /> "d",<br /> "e",<br /> "f"<br /> ]<br /> }<br />
  
  

  3. Pipeline设计
  

  
  

  如何管理Pipeline
  

  
  在IngestServcie中有一个Map存储所有的Pipeline实例，private volatile Map<String, PipelineHolder> pipelines = Collections.emptyMap();
  
  这里并没有将Pipeline实例存储在IngestMetadata,这样做的原因有2个：
  
  

  1. 在 Elasticsearch 的启动过程中，插件和节点服务的初始化发生在 ClusterState 加载之后, 只有等所有插件完成加载后，所有的Processor工厂才会被注册到系统中，这意味着在集群状态初始化之前，Processor工厂并不可用。
     
  2. ClusterState 中的元数据结构是静态注册的，即在类加载时已经确定，如果要将运行时示例存储进去，必须改变 ClusterState 的元数据结构存储格式，这个很不方便维护，而且这在集群状态同步的时候也会带来不必要的序列化以及反序列化。
     
     所以Pipeline的管理逻辑是：
     
     
• ClusterState 中的IngestMetadata 只存储 JSON 格式的管道定义，描述 Pipeline 的配置（例如，包含哪些 Processor，它们的参数等）。
  
• IngestService 中维护的运行时实例，将 JSON 配置解析为完整的 Pipeline 对象，包括处理器链
  
  在集群变更时， pipeline的CRUD的实现在innerUpdatePipelines方法中 。
  
  

  责任链设计模式的应用
  

  
  管道的执行通过IngestDocument的org.elasticsearch.ingest.IngestDocument#executePipeline 去驱动，每个文档的每个Pipeline都会进入到这个函数， 而每个Pipeline有组装好的CompoundProcessor，实际的链式调用是在CompoundProcessor中。
  
  简图大致如下:
  
  

  

  
  
  
  
  这里拿其和Zookeeper(3.6.3)中RequestProcessor 做一些对比：
  
  | 特点 | ES中的CompoundProcessor | Zookeeper中RequestProcessor |
  | --- | --- | --- |
  | 链的存储定义 | 由两条列表组成，分别保存正常流程以及失败流程的Processor列表。 | 固定的链式结构。 |
  | 异步执行 | 支持异步回调，可以异步执行链中的处理器。 | 同步执行 。 |
  | 失败处理 | 提供专门的 onFailureProcessors 作为失败处理链。如果不忽略异常并且onFailureProcessors 不为空则会执行失败处理链逻辑。 | 没有专门的失败处理链，异常直接交由上层捕获。 |
  | 可配置性 | 可动态调整处理器链和配置（如 ignoreFailure）。 | 代码中写死，无法配置 |
  
  这里并没有说Zookeeper的设计就差于Elasticsearch， 只是设计目标有所不同，RequestProcessor就是适合集中式的强一致、其中Processor并不需要灵活变化，而CompoundProcessor就是适合高并发而Procesor灵活变化场景。
  
  

  4. Ingest实战建议
  

  
  回到实战中来，这里结合目前所分析的内容给出相应的实战建议。
  
  

• 建立监控
  
  • 对于关键的Piepline，我们需要通过GET _nodes/stats?filter_path=nodes.*.ingest 获取其运行状况，识别延迟或失败的 Processor。
    
  • 文档写入的时候使用的是write线程池，我们也需要监控GET _nodes/stats?filter_path=nodes.*.thread_pool.write 的queue和write_rejections 判断需要扩展线程池。
    
• 优化Processor
  
  • 减少高开销操作，优先使用内置 Processor，比如script Processor 可适时替换set， append，避免过度复杂的正则表达式或嵌套逻辑。
    
  • 自定义的Processor尽量优化，比如如果涉及查询外部系统可考虑引入缓存。
    
• 建立单独的Ingest节点
  
  • 如果有大量的Pipeline需要执行，则可以考虑增加专用 Ingest 节点，避免与数据节点争夺资源。
    
    

    5. 漏洞&&修复分析
    

    
    这里分析7.10.2版本Ingest模块存在的漏洞以及官方是如何修复的。
    
    

    CVE-2021-22144
    

    
    https://discuss.elastic.co/t/e ... 78100
    
    Elasticsearch Grok 解析器中发现了一个不受控制的递归漏洞，该漏洞可能导致拒绝服务攻击。能够向 Elasticsearch 提交任意查询的用户可能会创建恶意 Grok 查询，从而导致 Elasticsearch 节点崩溃。
    
    

    漏洞复现
    

    
    发起这个请求：
    
    

• patterns: 处理字段时使用的 Grok 模式，这里设置为 %{INT}。
  
• pattern_definitions: 定义自定义 Grok 模式，这里故意让 INT 模式递归引用自身，导致循环引用问题。
  
  go<br /> POST /_ingest/pipeline/_simulate<br /> {<br /> "pipeline": {<br /> "processors": [<br /> {<br /> "grok": {<br /> "field": "message",<br /> "patterns": [<br /> "%{INT}"<br /> ],<br /> "pattern_definitions": {<br /> "INT": "%{INT}"<br /> }<br /> }<br /> }<br /> ]<br /> },<br /> "docs": [<br /> {<br /> "_source": {<br /> "message": "test"<br /> }<br /> }<br /> ]<br /> }<br /> <br />
  
  当执行之后会使得节点直接StackOverflow中止进程。
  
  

  修复逻辑
  

  
  这个问题的关键在于原先的逻辑中，只会对间接的递归引用（pattern1 => pattern2 => pattern3 => pattern1）做了检测，但是没有对直接的自引用(pattern1 => pattern1 )做检测。
  
  java<br /> private void forbidCircularReferences() {<br /> // 这个是增加的逻辑，检测直接的自引用<br /> for (Map.Entry<String, String> entry : patternBank.entrySet()) {<br /> if (patternReferencesItself(entry.getValue(), entry.getKey())) {<br /> throw new IllegalArgumentException("circular reference in pattern [" + entry.getKey() + "][" + entry.getValue() + "]");<br /> }<br /> }<br /> <br /> // 间接递归引用检测（这个是原先的逻辑）<br /> for (Map.Entry<String, String> entry : patternBank.entrySet()) {<br /> String name = entry.getKey();<br /> String pattern = entry.getValue();<br /> innerForbidCircularReferences(name, new ArrayList<>(), pattern);<br /> }<br /> }<br />
  
  

  CVE-2023-46673
  

  
  https://discuss.elastic.co/t/e ... 47708
  
  

  漏洞复现
  

  
  尝试了很多已有的Processor都没有复现，我们这使用自定义的Processor来复现，将之前的自定义AddArrayProcessor加一行代码：
  
  java<br /> @Override<br /> public IngestDocument execute(IngestDocument ingestDocument) throws Exception {<br /> List valueList = new ArrayList<>(Arrays.asList(value.split(",")));<br /> valueList.add(valueList); // 增加的代码<br /> ingestDocument.setFieldValue(field, valueList);<br /> return ingestDocument;<br /> }<br />
  
  重新编译再安装插件之后，执行改Processor 将会StackOverflow。
  
  

  修复逻辑
  

  
  这个问题的关键在于在IngestDocument的deepCopyMap的方法之前没有判断这样的无限引用的情况：
  
  

  

  
  
  
  
  那么在此之前做一个检测就好了，这个方法在原本的ES代码中就存在：org.elasticsearch.common.util.CollectionUtils#ensureNoSelfReferences(java.lang.Object, java.lang.String) ，其利用 IdentityHashMap 记录已访问对象的引用，检测并防止对象间的循环引用。
  
  

  CVE-2024-23450
  

  
  https://discuss.elastic.co/t/e ... 56314
  
  

  漏洞复现
  

  
  虽然我们的索引只有2个Pipeline的配置，但是由于Pipeline Processor的存在，所以实际上一个文档其实能被很多Pipeline处理，当需要执行足够多个的pipline个数时，则会发生StackOverflow。
  
  

  修复逻辑
  

  
  这个问题的关键在于对Pipeline的个数并没有限制，添加一个配置项，当超出该个数则直接抛出异常。
  
  java<br /> public static final int MAX_PIPELINES = Integer.parseInt(System.getProperty("es.ingest.max_pipelines", "100"));<br />
  
  IngestDocument的org.elasticsearch.ingest.IngestDocument#executePipeline 添加逻辑：
  
  java<br /> public void executePipeline(Pipeline pipeline, BiConsumer<IngestDocument, Exception> handler) {<br /> if (executedPipelines.size() >= MAX_PIPELINES) {<br /> handler.accept(<br /> null,<br /> new IllegalStateException(PIPELINE_TOO_MANY_ERROR_MESSAGE + MAX_PIPELINES + " nested pipelines")<br /> );<br /> } <br />
  
  思考： 这里判断pipeline是否超出100个限制是用已经执行的pipeline个数来计算的。 假设已经超出100个pipeline，那这100个pipeline是会白跑的， 如果能在真正执行之前分析需要执行的Pipeline个数会更好。
  
  

  6. 总结
  

  
  Ingest 模块作为 Elasticsearch 数据处理流程的重要组成部分，提供了灵活的管道化能力，使得用户能够在数据写入前进行丰富的预处理操作。然而，在实际场景中，Ingest 模块也面临性能瓶颈、资源竞争等挑战，需要结合业务需求和系统现状进行精细化调优，通过优化 Processor 执行效率、合理规划集群架构以及增强监控与诊断手段，我们可以充分释放 Ingest 模块的能力，提升 Elasticsearch 的整体数据处理能力。
  

今日快讯！就在刚刚，开源搜索领域行业巨头 Elastic 官方博客发表了一篇最新公告《Elasticsearch is Open Source, Again》，Elastic 创始人& CTO Shay Banon 宣布 Elasticsearch 和 Kibana 许可证协议修改为 AGPL。

以下为搜索客社区从 Elastic 官方博客翻译的原文内容：

---

Elasticsearch 再次成为开源软件

[D.N.A] Elasticsearch 和 Kibana 现在可以再次被称为开源软件了。这句话让我感到无比的兴奋。真的忍不住跳起来庆祝！我们所有在 Elastic 的人都很高兴。开源精神是我的DNA，也是 Elastic 的DNA。能够再次称 Elasticsearch 为开源软件，真的是一种纯粹的快乐。


[LOVE.] 简而言之，我们将在接下来的几周内，除了 ELv2 和 SSPL 之外，增加 AGPL 作为另一个许可选项。在更改许可后，我们从未停止过像一个开源社区一样相信和行动。但通过使用 AGPL 这一获得开放源码促进会（OSI）批准的许可，消除了人们可能存在的任何疑问或混淆。


[Not Like Us] 我们在 Elastic 从未停止过对开源的信仰。我个人对开源的信仰也从未动摇，至今已有 25 年了。那么为什么三年前我们要做出改变呢？我们遇到了与 AWS 相关的问题，以及他们的服务引发的市场混乱。在尝试了所有能想到的其他选项后，我们更改了许可协议，明知道这会导致 Elasticsearch 被分叉成另一个名称并走上不同的发展轨迹。这是一个漫长的故事。


[Like That] 好消息是，虽然过程痛苦，但结果奏效了。三年后，亚马逊已经完全投入到了他们的分叉项目中，市场的混乱（大部分）得到了缓解，我们与 AWS 的合作伙伴关系比以往更强。我们甚至被评为 AWS 的年度合作伙伴。我一直希望时间能过去得足够久，以至于我们可以安全地回到开源项目的状态——现在终于到了。


[All The Stars] 我们希望尽可能简化用户的使用体验。我们有用户非常喜欢 ELv2（一个受 BSD 启发的许可）。我们也有用户已经批准使用 SSPL（通过 MongoDB 使用）。这就是为什么我们只是增加了一个选项，而不是移除任何东西。如果你已经在使用并喜欢 Elasticsearch，请继续使用，没有任何变化。对于其他人，现在你也可以选择使用 AGPL。


[LOYALTY.] 我们选择 AGPL 而不是其他许可，是因为我们希望通过与 OSI 的合作，能在开源许可领域创造更多的选项。随着我们更改许可以来的发展（例如 Grafana 从 Apache2 转移到 AGPL），也许 AGPL 已经足够适用于像我们这样的基础设施软件了。我们致力于找到最合适的解决方案。


[euphoria] 我非常高兴能够再次称 Elasticsearch 为开源软件。


[Alright] 任何改变都可能引发混淆，当然也可能引来一些网络喷子。（网络喷子总是存在的，对吧？）让我们愉快地回答一些可能出现的问题吧。我可以想象到的一些问题如下，但我们会继续补充。


“更改许可是个错误，Elastic 现在在回撤”：我们在三年前更改许可时消除了很多市场混乱。由于我们的行动，很多事情已经改变。现在是一个完全不同的市场环境。我们不生活在过去。我们希望为用户打造一个更好的未来。正是因为我们当时采取了行动，现在我们才有能力采取新的行动。


“AGPL 不是真正的开源许可，X 才是”：AGPL 是获得 OSI 批准的许可，并且被广泛采用。例如，MongoDB 曾经使用 AGPL，Grafana 现在也是 AGPL。这表明 AGPL 并不影响使用或流行度。我们选择 AGPL 是因为我们认为这是与 OSI 一起为世界开辟更多开源道路的最佳方式，而不是减少开源。


“Elastic 更改许可是因为他们表现不好”：我要说的是，我今天对 Elastic 的未来依然充满期待。我为我们的产品和团队的执行力感到无比自豪。我们发布了无状态的 Elasticsearch ES|QL 和大量用于 GenAI 用例的向量数据库/混合搜索改进。我们在日志记录和可观察性方面大力投入 OTel。我们的安全产品 SIEM 不断添加令人惊叹的功能，并且是市场上增长最快的产品之一。用户的反应让我们感到非常谦卑。股市总有起伏，但我可以向你保证，我们始终着眼于长远发展，而这次的变更就是其中的一部分。


如果我们看到更多问题，会在上面继续添加，以期减少混淆。


[HUMBLE.] 为未来而构建真是令人兴奋。Elasticsearch 回归开源。万岁！这真是一件美妙的事情。今天真是美好的一天。


Forever :elasticheart: Open Source

Shay 2024-08-30

---

原文地址：https://www.elastic.co/blog/el ... gain/

社区热评

Elasticsearch 再次回归开源的消息迅速引发了技术社区的广泛关注。这不仅是 Elastic 对自身开源信念的重申，也是其在激烈市场竞争中精心策划的一次战略调整。

三年前，Elastic 因与 AWS 的市场竞争而选择更改许可协议，这一决定在当时引发了不小的争议。尽管如此，事实证明，这一策略有效地减少了市场上的混淆，也为 Elastic 与 AWS 的合作奠定了更坚实的基础。如今，Elastic 再度选择开源，并新增 AGPL 作为许可选项，这一举措无疑展现了 Elastic 在市场中更加成熟的定位和对未来发展的自信。

这一变化不仅仅是一个公司的商业决策，更是开源生态系统的一次重要信号。Elastic 的回归开源，传递出一个明确的信息：即使在商业竞争中，开源仍然是企业实现长远发展的重要路径。随着这一决定的落地，其他软件公司可能也会重新审视自身的许可策略，推动更多开源项目的发展与创新。

此外，Elastic 选择 AGPL 作为新许可选项，也显示出其对开源生态未来走向的深刻洞察。AGPL 的引入，表明 Elastic 希望在开源社区中保持灵活性和多样性，同时推动整个行业向更加开放和透明的方向迈进。

总体而言，Elastic 重返开源的举动，不仅意在巩固其在开源社区中的地位，也为行业树立了一个新的标杆。这一事件无疑将成为开源软件发展史上的重要篇章，未来可能会激发更多企业重新考虑其开源战略，进而推动整个技术行业的进一步繁荣与进步。

让我们拭目以待！

本次活动由 搜索客社区、极限科技（INFINI Labs）联合举办，活动主题将深入探讨 Elasticsearch 的两个核心方面：代码结构以及写入和查询的关键流程。本次活动将为 Elasticsearch 初学者和有经验的用户提供宝贵的见解，欢迎大家报名参加、交流学习。

活动主题：Elasticsearch 的代码结构和写入查询流程的解读 - 下篇
活动时间：2024 年 8 月 28 日 19:00-20:00（周三）
活动形式：微信视频号（极限实验室）直播
报名方式：关注或扫码海报中的二维码进行预约


嘉宾介绍

张磊，极限科技搜索引擎研发负责人，对 Elasticsearch 和 Lucene 源码比较熟悉，目前主要负责公司的 Easysearch 产品的研发以及客户服务工作。

主题摘要

本次分享将探讨 Elasticsearch 的代码结构及其写入和查询流程。内容包括：项目架构、核心模块、插件系统，以及文档索引和查询的各个阶段与分布式查询协调。

活动亮点

1. 深入解析 Elasticsearch 代码结构：
   
   
   • 项目的整体结构：了解 Elasticsearch 项目的组织方式。
     
   • 核心模块及其职责：学习 Elasticsearch 的核心模块及其在系统中的具体角色。
     
   • 插件系统的设计：探索 Elasticsearch 灵活的插件系统设计及其扩展方式。
     
     
2. 详细解读 Elasticsearch 写入和查询流程：
   
   
   • 文档索引过程的各个阶段：跟随文档从初始接收至最终存储的索引过程。
     
   • 查询解析和执行的步骤：理解 Elasticsearch 中查询解析和执行的各个步骤。
     
   • 分布式查询的协调过程：学习 Elasticsearch 如何在分布式架构中协调查询，以提供高效且准确的搜索结果。
     
     

     参与有奖
     

     
     直播中将设有福袋抽奖环节，参与就有机会获得 INFINI Labs 周边纪念品，包括 T 恤、鸭舌帽、咖啡杯、指甲刀套件等等（图片仅供参考，款式、颜色与尺码随机）。
     
     
     
     

     活动交流
     

     
     本活动设有 Meetup 技术交流群，可添加小助手微信入群。
     
     
     
     本次 Meetup 是深入了解 Elasticsearch 内部工作机制、提升使用技能的绝佳机会。不要错过！
     
     

     Meetup 讲师招募
     

     
     
     
     搜索客社区 Meetup 的成功举办，离不开社区小伙伴的热情参与。目前社区讲师招募计划也在持续进行中，我们诚挚邀请各位技术大咖、行业精英踊跃提交演讲议题，与大家分享您的经验。
     
     讲师报名链接：[http://cfp.searchkit.cn](http://cfp.searchkit.cn)
     或扫描下方二维码，立刻报名成为讲师！
     
     
     
     Meetup 活动聚焦 AI 与搜索领域的最新动态，以及数据实时搜索分析、向量检索、技术实践与案例分析、日志分析、安全等领域的深度探讨。
     
     我们热切期待您的精彩分享！
     
     

     往期回顾
     

     
     

   • [【第 2 期】2024 搜索客 Meetup | Elasticsearch 的代码结构和写入查询流程的解读 - 上篇](https://mp.weixin.qq.com/s/oQNej6aDMzLp64_AKxSONw)
     
   • [【第 1 期】2024 搜索客 Meetup | Easysearch 结合大模型实现 RAG](https://mp.weixin.qq.com/s/7fpPFGKltJASspmIaUqJMg)
     
     

     关于 搜索客（SearchKit）社区
     

     
     搜索客社区由 Elasticsearch 中文社区进行全新的品牌升级，以新的 Slogan：“搜索人自己的社区” 为宣言。汇集搜索领域最新动态、精选干货文章、精华讨论、文档资料、翻译与版本发布等，为广大搜索领域从业者提供更为丰富便捷的学习和交流平台。
     
     社区官网：[https://searchkit.cn](https://searchkit.cn) 。
     

故障现象

近日有客户找到我们，说有个 ES 集群节点，磁盘利用率达到了 82% ，而其节点才 63% ，想处理下这个节点，降低节点的磁盘利用率。

起初以为是没有打开自动平衡导致的，经查询，数据还是比较平衡的。
利用率较高的是 76 节点，如果 76 节点的分片比其他节点多，好像还比较合乎逻辑，但它反而比其他节点少了 12-15 个分片。那是 76 节点上的分片比较大？

索引情况

图中都是较大的索引，1 个索引 25TB 左右，共 160 个分片。

分片大小

节点 64
节点 77
节点 75
问题节点 76
可以看出分片大小没有出现较大的倾斜，分片大小和数据平衡的原因都被排除。

换个方向思考，节点 76 比其他节点多使用了磁盘空间 8 个 TB 左右，集群最大分片大小约 140GB ，8000/140=57 ，即节点 76 至少要比其他节点多 57 个分片才行，啊这...

会不会有其他的文件占用了磁盘空间？

我们登录到节点主机，排查是否有其他文件占用了磁盘空间。

结果：客户的数据路径是单独的数据磁盘，并没有其他文件，都是 ES 集群索引占用的空间。

现象总结

分片大小差不多的情况下，节点 76 的分片数还比别的节点还少 10 个左右，它的磁盘空间反而多占用了 8TB 。

这是不是太奇怪了？事出反常必有妖，继续往下查。

原因定位

通过进一步排查，我们发现节点 76 上有一批索引目录，在其他的节点上没有，而且也不在 GET \_cat/indices?v 命令的结果中。说明这些目录都是 dangling 索引占用的。

dangling 索引产生的原因

当 Elasticsearch 节点脱机时，如果删除的索引数量超过 Cluster.indes.tombstones.size，就会发生这种情况。

解决方案

通过命令删除 dangling 索引：

<br /> DELETE /\_dangling/<index-uuid>?accept_data_loss=true<br />

最后

这次的分享就到这里了，欢迎与我一起交流 ES 的各种问题和解决方案。

关于极限科技（INFINI Labs）


极限科技，全称极限数据（北京）科技有限公司，是一家专注于实时搜索与数据分析的软件公司。旗下品牌极限实验室（INFINI Labs）致力于打造极致易用的数据探索与分析体验。

极限科技是一支年轻的团队，采用天然分布式的方式来进行远程协作，员工分布在全球各地，希望通过努力成为中国乃至全球企业大数据实时搜索分析产品的首选，为中国技术品牌输出添砖加瓦。

官网：[https://infinilabs.cn](https://infinilabs.cn)

构建Elasticsearch专家Bot的详细步骤指南

步骤1: 文档搜集

• 利用专业工具搜集Elasticsearch 8.13.2版本的官方文档，确保文档的完整性和准确性。如：Elasticsearch文档
  
  

  步骤2: 知识库建立
  

• 在coze.cn平台上创建一个专门的知识库，命名清晰，便于管理和识别。
  
  

  步骤3: 文档上传
  

• 将搜集到的Elasticsearch文档上传至新创建的知识库，确保文档格式适合后续的检索和分析。
  
  

  步骤4: Bot创建
  

• 在coze.cn上创建一个新的Bot，命名为“Elasticsearch专家”，为其设定一个专业且引人注目的形象。
  
  

  步骤5: 知识库配置
  

• 将步骤2中的知识库与新创建的Bot进行关联，确保Bot能够访问和利用这些文档资源。
  
  

  步骤6: 功能插件集成
  

• 为Bot添加以下功能插件，以提供更全面的服务：
  
  • 必应搜索引擎（Bing Web Search）：扩展信息检索范围。
    
  • 代码执行器（CodeRunner）：实现代码的即时测试与验证。
    
  • 微信搜索（WeChat Search）：增加中文信息源的覆盖。
    
    

    步骤7: 人设与回复逻辑定制
    

• 设定Bot的人设，明确其专业领域和能力，如：“我是Elasticsearch的专家，随时准备解答你的疑问。”
  
• 利用coze平台的AI技术，优化Bot的回复逻辑，确保其回答既准确又具有针对性。
  
  

  步骤8: 测试与调整
  

• 在Bot设置完成后，进行全面的测试，确保其能够正确理解和回应各种查询。
  
• 根据测试反馈，调整Bot的交互逻辑和回答内容，以提高用户满意度。
  
  

  步骤9: 发布与分享
  

• 完成所有设置和测试后，点击发布，使Bot正式上线。
  
• 通过Bot页面的商店功能，将你的“Elasticsearch专家”Bot分享给你的伙伴们，让他们也能享受到这一强大的学习工具。
  
  

  步骤10: 持续优化与更新
  

• 定期回顾Bot的表现，根据用户反馈进行持续的优化和功能更新。
  
• 随着Elasticsearch版本的迭代，及时更新知识库内容，确保Bot提供的信息始终最新。
  
  通过遵循这些步骤，你不仅能够构建一个功能全面的Elasticsearch专家Bot，而且能够确保它随着时间的推移不断进化，满足用户日益增长的需求。这将是一个不仅能提供文档查询，还能执行代码和搜索网络的智能助手，极大地提升你的Elasticsearch学习之旅。