行动是治愈恐惧的良药，而犹豫、拖延将不断滋养恐惧。
gateway

gateway

ES 调优帖：Gateway 批量写入性能优化实践

Elasticsearch • INFINI Labs 小助手 发表了文章 • 0 个评论 • 5006 次浏览 • 2025-08-06 17:32 • 来自相关话题

背景：bulk 优化的应用

在 ES 的写入优化里，bulk 操作被广泛地用于批量处理数据。bulk 操作允许用户一次提交多个数据操作，如索引、更新、删除等，从而提高数据处理效率。bulk 操作的实现原理是，将数据操作请求打包成 HTTP 请求，并批量提交给 Elasticsearch 服务器。这样，Elasticsearch 服务器就可以一次处理多个数据操作，从而提高处理效率。

这种优化的核心价值在于减少了网络往返的次数和连接建立的开销。每一次单独的写入操作都需要经历完整的请求-响应周期，而批量写入则是将多个操作打包在一起，用一次通信完成原本需要多次交互的工作。这不仅仅节省了时间，更重要的是释放了系统资源，让服务器能够专注于真正的数据处理，而不是频繁的协议握手和状态维护。

这样的批量请求的确是可以优化写入请求的效率，让 ES 集群获得更多的资源去做写入请求的集中处理。但是除了客户端与 ES 集群的通讯效率优化，还有其他中间过程能优化么？

Gateway 的优化点

bulk 的优化理念是将日常零散的写入需求做集中化的处理，尽量减低日常请求的损耗，完成资源最大化的利用。简而言之就是“好钢用在刀刃上”。

但是 ES 在收到 bulk 写入请求后，也是需要协调节点根据文档的 id 计算所属的分片来将数据分发到对应的数据节点的。这个过程也是有一定损耗的，如果 bulk 请求中数据分布的很散，每个分片都需要进行写入，原本 bulk 集中写入的需求优势则还是没有得到最理想化的提升。

gateway 的写入加速则对 bulk 的优化理念的最大化补全。

gateway 可以本地计算每个索引文档对应后端 Elasticsearch 集群的目标存放位置，从而能够精准的进行写入请求定位。

在一批 bulk 请求中，可能存在多个后端节点的数据，bulk_reshuffle 过滤器用来将正常的 bulk 请求打散，按照目标节点或者分片进行拆分重新组装，避免 Elasticsearch 节点收到请求之后再次进行请求分发， 从而降低 Elasticsearch 集群间的流量和负载，也能避免单个节点成为热点瓶颈，确保各个数据节点的处理均衡，从而提升集群总体的索引吞吐能力。

整理的优化思路如下图：

优化实践

那我们来实践一下，看看 gateway 能提升多少的写入。

这里我们分 2 个测试场景：

  1. 基础集中写入测试，不带文档 id，直接批量写入。这个场景更像是日志或者监控数据采集的场景。
  2. 带文档 id 的写入测试，更偏向搜索场景或者大数据批同步的场景。

2 个场景都进行直接写入 ES 和 gateway 转发 ES 的效率比对。

测试材料除了需要备一个网关和一套 es 外，其余的内容如下：

测试索引 mapping 一致，名称区分：

PUT gateway_bulk_test
{
  "settings": {
    "number_of_shards": 6,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "timestamp": {
        "type": "date",
        "format": "strict_date_optional_time"
      },
      "field1": {
        "type": "keyword"
      },
      "field2": {
        "type": "keyword"
      },
      "field3": {
        "type": "keyword"
      },
      "field4": {
        "type": "integer"
      },
      "field5": {
        "type": "keyword"
      },
      "field6": {
        "type": "float"
      }
    }
  }
}

PUT bulk_test
{
  "settings": {
    "number_of_shards": 6,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "timestamp": {
        "type": "date",
        "format": "strict_date_optional_time"
      },
      "field1": {
        "type": "keyword"
      },
      "field2": {
        "type": "keyword"
      },
      "field3": {
        "type": "keyword"
      },
      "field4": {
        "type": "integer"
      },
      "field5": {
        "type": "keyword"
      },
      "field6": {
        "type": "float"
      }
    }
  }
}

gateway 的配置文件如下：

path.data: data
path.logs: log

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 200000
    network:
      binding: 0.0.0.0:8000

flow:
  - name: async_bulk
    filter:
      - bulk_reshuffle:
          when:
            contains:
              _ctx.request.path: /_bulk
          elasticsearch: prod
          level: node
          partition_size: 1
          fix_null_id: true
      - elasticsearch:
          elasticsearch: prod #elasticsearch configure reference name
          max_connection_per_node: 1000 #max tcp connection to upstream, default for all nodes
          max_response_size: -1 #default for all nodes
          balancer: weight
          refresh: # refresh upstream nodes list, need to enable this feature to use elasticsearch nodes auto discovery
            enabled: true
            interval: 60s
          filter:
            roles:
              exclude:
                - master

router:
  - name: my_router
    default_flow: async_bulk

elasticsearch:
  - name: prod
    enabled: true
    endpoints:
      - https://127.0.0.1:9221
      - https://127.0.0.1:9222
      - https://127.0.0.1:9223
    basic_auth:
      username: admin
      password: admin

pipeline:
  - name: bulk_request_ingest
    auto_start: true
    keep_running: true
    retry_delay_in_ms: 1000
    processor:
      - bulk_indexing:
          max_connection_per_node: 100
          num_of_slices: 3
          max_worker_size: 30
          idle_timeout_in_seconds: 10
          bulk:
            compress: false
            batch_size_in_mb: 10
            batch_size_in_docs: 10000
          consumer:
            fetch_max_messages: 100
          queue_selector:
            labels:
              type: bulk_reshuffle

测试脚本如下：

#!/usr/bin/env python3
"""
ES Bulk写入性能测试脚本

"""

import hashlib
import json
import random
import string
import time
from typing import List, Dict, Any

import requests
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
import urllib3

# 禁用SSL警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class ESBulkTester:
    def __init__(self):
        # 配置变量 - 可修改
        self.es_configs = [
            {
                "name": "ES直连",
                "url": "https://127.0.0.1:9221",
                "index": "bulk_test",
                "username": "admin",  # 修改为实际用户名
                "password": "admin",  # 修改为实际密码
                "verify_ssl": False  # HTTPS需要SSL验证
            },
            {
                "name": "Gateway代理",
                "url": "http://localhost:8000",
                "index": "gateway_bulk_test",
                "username": None,  # 无需认证
                "password": None,
                "verify_ssl": False
            }
        ]
        self.batch_size = 10000  # 每次bulk写入条数
        self.log_interval = 100000  # 每多少次bulk写入输出日志

        # ID生成规则配置 - 前2位后5位
        self.id_prefix_start = 1
        self.id_prefix_end = 999      # 前3位: 01-999
        self.id_suffix_start = 1
        self.id_suffix_end = 9999   # 后4位: 0001-9999

        # 当前ID计数器
        self.current_prefix = self.id_prefix_start
        self.current_suffix = self.id_suffix_start

    def generate_id(self) -> str:
        """生成固定规则的ID - 前2位后5位"""
        id_str = f"{self.current_prefix:02d}{self.current_suffix:05d}"

        # 更新计数器
        self.current_suffix += 1
        if self.current_suffix > self.id_suffix_end:
            self.current_suffix = self.id_suffix_start
            self.current_prefix += 1
            if self.current_prefix > self.id_prefix_end:
                self.current_prefix = self.id_prefix_start

        return id_str

    def generate_random_hash(self, length: int = 32) -> str:
        """生成随机hash值"""
        random_string = ''.join(random.choices(string.ascii_letters + string.digits, k=length))
        return hashlib.md5(random_string.encode()).hexdigest()

    def generate_document(self) -> Dict[str, Any]:
        """生成随机文档内容"""
        return {
            "timestamp": datetime.now().isoformat(),
            "field1": self.generate_random_hash(),
            "field2": self.generate_random_hash(),
            "field3": self.generate_random_hash(),
            "field4": random.randint(1, 1000),
            "field5": random.choice(["A", "B", "C", "D"]),
            "field6": random.uniform(0.1, 100.0)
        }

    def create_bulk_payload(self, index_name: str) -> str:
        """创建bulk写入payload"""
        bulk_data = []

        for _ in range(self.batch_size):
            #doc_id = self.generate_id()
            doc = self.generate_document()

            # 添加index操作
            bulk_data.append(json.dumps({
                "index": {
                    "_index": index_name,
            #        "_id": doc_id
                }
            }))
            bulk_data.append(json.dumps(doc))

        return "\n".join(bulk_data) + "\n"

    def bulk_index(self, config: Dict[str, Any], payload: str) -> bool:
        """执行bulk写入"""
        url = f"{config['url']}/_bulk"
        headers = {
            "Content-Type": "application/x-ndjson"
        }

        # 设置认证信息
        auth = None
        if config.get('username') and config.get('password'):
            auth = (config['username'], config['password'])

        try:
            response = requests.post(
                url,
                data=payload,
                headers=headers,
                auth=auth,
                verify=config.get('verify_ssl', True),
                timeout=30
            )
            return response.status_code == 200
        except Exception as e:
            print(f"Bulk写入失败: {e}")
            return False

    def refresh_index(self, config: Dict[str, Any]) -> bool:
        """刷新索引"""
        url = f"{config['url']}/{config['index']}/_refresh"

        # 设置认证信息
        auth = None
        if config.get('username') and config.get('password'):
            auth = (config['username'], config['password'])

        try:
            response = requests.post(
                url,
                auth=auth,
                verify=config.get('verify_ssl', True),
                timeout=10
            )
            success = response.status_code == 200
            print(f"索引刷新{'成功' if success else '失败'}: {config['index']}")
            return success
        except Exception as e:
            print(f"索引刷新失败: {e}")
            return False

    def run_test(self, config: Dict[str, Any], round_num: int, total_iterations: int = 100000):
        """运行性能测试"""
        test_name = f"{config['name']}-第{round_num}轮"
        print(f"\n开始测试: {test_name}")
        print(f"ES地址: {config['url']}")
        print(f"索引名称: {config['index']}")
        print(f"认证: {'是' if config.get('username') else '否'}")
        print(f"每次bulk写入: {self.batch_size}条")
        print(f"总计划写入: {total_iterations * self.batch_size}条")
        print("-" * 50)

        start_time = time.time()
        success_count = 0
        error_count = 0

        for i in range(1, total_iterations + 1):
            payload = self.create_bulk_payload(config['index'])

            if self.bulk_index(config, payload):
                success_count += 1
            else:
                error_count += 1

            # 每N次输出日志
            if i % self.log_interval == 0:
                elapsed_time = time.time() - start_time
                rate = i / elapsed_time if elapsed_time > 0 else 0
                print(f"已完成 {i:,} 次bulk写入, 耗时: {elapsed_time:.2f}秒, 速率: {rate:.2f} bulk/秒")

        end_time = time.time()
        total_time = end_time - start_time
        total_docs = total_iterations * self.batch_size

        print(f"\n{test_name} 测试完成!")
        print(f"总耗时: {total_time:.2f}秒")
        print(f"成功bulk写入: {success_count:,}次")
        print(f"失败bulk写入: {error_count:,}次")
        print(f"总文档数: {total_docs:,}条")
        print(f"平均速率: {success_count/total_time:.2f} bulk/秒")
        print(f"文档写入速率: {total_docs/total_time:.2f} docs/秒")
        print("=" * 60)

        return {
            "test_name": test_name,
            "config_name": config['name'],
            "round": round_num,
            "es_url": config['url'],
            "index": config['index'],
            "total_time": total_time,
            "success_count": success_count,
            "error_count": error_count,
            "total_docs": total_docs,
            "bulk_rate": success_count/total_time,
            "doc_rate": total_docs/total_time
        }

    def run_comparison_test(self, total_iterations: int = 10000):
        """运行双地址对比测试"""
        print("ES Bulk写入性能测试开始")
        print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("=" * 60)

        results = []
        rounds = 2  # 每个地址测试2轮

        # 循环测试所有配置
        for config in self.es_configs:
            print(f"\n开始测试配置: {config['name']}")
            print("*" * 40)

            for round_num in range(1, rounds + 1):
                # 运行测试
                result = self.run_test(config, round_num, total_iterations)
                results.append(result)

                # 每轮结束后刷新索引
                print(f"\n第{round_num}轮测试完成，执行索引刷新...")
                self.refresh_index(config)

                # 重置ID计数器
                if round_num == 1:
                    # 第1轮：使用初始ID范围（新增数据）
                    print("第1轮：新增数据模式")
                else:
                    # 第2轮：重复使用相同ID（更新数据模式）
                    print("第2轮：数据更新模式，复用第1轮ID")
                    self.current_prefix = self.id_prefix_start
                    self.current_suffix = self.id_suffix_start

                print(f"{config['name']} 第{round_num}轮测试结束\n")

        # 输出对比结果
        print("\n性能对比结果:")
        print("=" * 80)

        # 按配置分组显示结果
        config_results = {}
        for result in results:
            config_name = result['config_name']
            if config_name not in config_results:
                config_results[config_name] = []
            config_results[config_name].append(result)

        for config_name, rounds_data in config_results.items():
            print(f"\n{config_name}:")
            total_time = 0
            total_bulk_rate = 0
            total_doc_rate = 0

            for round_data in rounds_data:
                print(f"  第{round_data['round']}轮:")
                print(f"    耗时: {round_data['total_time']:.2f}秒")
                print(f"    Bulk速率: {round_data['bulk_rate']:.2f} bulk/秒")
                print(f"    文档速率: {round_data['doc_rate']:.2f} docs/秒")
                print(f"    成功率: {round_data['success_count']/(round_data['success_count']+round_data['error_count'])*100:.2f}%")

                total_time += round_data['total_time']
                total_bulk_rate += round_data['bulk_rate']
                total_doc_rate += round_data['doc_rate']

            avg_bulk_rate = total_bulk_rate / len(rounds_data)
            avg_doc_rate = total_doc_rate / len(rounds_data)

            print(f"  平均性能:")
            print(f"    总耗时: {total_time:.2f}秒")
            print(f"    平均Bulk速率: {avg_bulk_rate:.2f} bulk/秒")
            print(f"    平均文档速率: {avg_doc_rate:.2f} docs/秒")

        # 整体对比
        if len(config_results) >= 2:
            config_names = list(config_results.keys())
            config1_avg = sum([r['bulk_rate'] for r in config_results[config_names[0]]]) / len(config_results[config_names[0]])
            config2_avg = sum([r['bulk_rate'] for r in config_results[config_names[1]]]) / len(config_results[config_names[1]])

            if config1_avg > config2_avg:
                faster = config_names[0]
                rate_diff = config1_avg - config2_avg
            else:
                faster = config_names[1]
                rate_diff = config2_avg - config1_avg

            print(f"\n整体性能对比:")
            print(f"{faster} 平均性能更好，bulk速率高 {rate_diff:.2f} bulk/秒")
            print(f"性能提升: {(rate_diff/min(config1_avg, config2_avg)*100):.1f}%")

def main():
    """主函数"""
    tester = ESBulkTester()

    # 运行测试（每次bulk 1万条，300次bulk = 300万条文档）
    tester.run_comparison_test(total_iterations=300)

if __name__ == "__main__":
    main()

1. 日志场景：不带 id 写入

测试条件：

  1. bulk 写入数据不带文档 id
  2. 每批次 bulk 10000 条数据，总共写入 30w 数据

这里把

反馈结果：

性能对比结果:
================================================================================

ES直连:
  第1轮:
    耗时: 152.29秒
    Bulk速率: 1.97 bulk/秒
    文档速率: 19699.59 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 152.29秒
    平均Bulk速率: 1.97 bulk/秒
    平均文档速率: 19699.59 docs/秒

Gateway代理:
  第1轮:
    耗时: 115.63秒
    Bulk速率: 2.59 bulk/秒
    文档速率: 25944.35 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 115.63秒
    平均Bulk速率: 2.59 bulk/秒
    平均文档速率: 25944.35 docs/秒

整体性能对比:
Gateway代理 平均性能更好，bulk速率高 0.62 bulk/秒
性能提升: 31.7%

2. 业务场景：带文档 id 的写入

测试条件：

  1. bulk 写入数据带有文档 id，两次测试写入的文档 id 生成规则一致且重复。
  2. 每批次 bulk 10000 条数据，总共写入 30w 数据

这里把 py 脚本中 第 99 行 和 第 107 行的注释打开。

反馈结果：

性能对比结果:
================================================================================

ES直连:
  第1轮:
    耗时: 155.30秒
    Bulk速率: 1.93 bulk/秒
    文档速率: 19317.39 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 155.30秒
    平均Bulk速率: 1.93 bulk/秒
    平均文档速率: 19317.39 docs/秒

Gateway代理:
  第1轮:
    耗时: 116.73秒
    Bulk速率: 2.57 bulk/秒
    文档速率: 25700.06 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 116.73秒
    平均Bulk速率: 2.57 bulk/秒
    平均文档速率: 25700.06 docs/秒

整体性能对比:
Gateway代理 平均性能更好，bulk速率高 0.64 bulk/秒
性能提升: 33.0%

小结

不管是日志场景还是业务价值更重要的大数据或者搜索数据同步场景， gateway 的写入加速都能平稳的节省 25%-30% 的写入耗时。

关于极限网关（INFINI Gateway）

INFINI Gateway 是一个开源的面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway，可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

官网文档：https://docs.infinilabs.com/gateway
开源地址：https://github.com/infinilabs/gateway

作者：金多安，极限科技（INFINI Labs）搜索运维专家，Elastic 认证专家，搜索客社区日报责任编辑。一直从事与搜索运维相关的工作，日常会去挖掘 ES / Lucene 方向的搜索技术原理，保持搜索相关技术发展的关注。
原文：https://infinilabs.cn/blog/2025/gateway-bulk-write-performance-optimization/

INFINI Labs 产品更新 | Coco AI – 增强 AI 搜索、API 管理与性能优化等

资讯动态 • INFINI Labs 小助手 发表了文章 • 0 个评论 • 3419 次浏览 • 2025-03-15 18:21 • 来自相关话题

release

INFINI Labs 产品更新发布！此次更新涵盖 Coco AI 、Easysearch 等产品多项重要升级，重点提升 AI 搜索能力、易用性及企业级优化。

  • Coco AI v0.2 作为 开源、跨平台的 AI 搜索工具，新增 APP 自动更新提示、API Token 管理、文档处理优化 等功能。
  • INFINI Easysearch v1.11.1 集成 AI 向量搜索，优化查询聚合能力。
  • INFINI Console、Gateway、Agent、Loadgen、Framework 关键问题修复，优化 WebSocket 处理与整体用户体验。
  • Coco AI Server 增强 WebSocket 会话处理、支持 RAG 会话和动态配置，并新增图形化管理界面。

Coco AI v0.2

Coco AI 是一个完全开源、跨平台的统一 AI 搜索与效率工具，能够连接并搜索多种数据源，包括应用程序、文件、谷歌网盘、Notion、语雀、Hugo 等本地与云端数据。通过接入 DeepSeek 等大模型，Coco AI 实现了智能化的个人知识库管理，注重隐私，支持私有部署，帮助用户快速、智能地访问信息。

Coco AI 本次详细更新记录如下：

Coco AI 客户端 v0.2.1

功能更新

  • 支持 APP 应用内更新提示并可自动更新

问题修复

  • 修复融合搜索包含已禁用服务器的问题
  • 修复版本类型不正确：应为字符串而不是 u32
  • 修复聊天推送结束的判断类型不准确问题

优化改进

  • 重构了聊天组件
  • 添加服务链接展示
  • 优化了聊天滚动效果和聊天数据渲染效果
  • 设置聊天窗口最小宽度 & 移除输入框背景
  • 移除废弃的选中功能 & 添加选择隐藏 APP 功能
  • Websocket 超时增加到 2 分钟

Coco AI 服务端 v0.2.2

功能更新

  • 新增图形化管理界面
  • 新增数据源下文档创建 API
  • 新增文件上传相关 API
  • 新增 API TOKEN 管理相关 API
  • 数据源同步支持动态配置时间间隔
  • 支持动态更新服务端设置
  • 支持动态更新大模型相关设置
  • 新增 RAG 聊天会话处理
  • 新增联网搜索能力
  • 支持对接 Deepseek 大模型
  • 新增文档预处理 Processor

问题修复

  • 修复 Google Drive Connector 缺少文件报错

优化改进

  • 优化聊天会话功能
  • 优化 Websocket 会话管理
  • 优化登录退出接口
  • 保存 Notion 其它内容到 Payload 字段
  • 完善后台任务退出机制
  • 优化默认索引模版和查询模版

INFINI Easysearch v1.11.1

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

Easysearch 本次更新如下：

功能更新

  • 新增 AI 模块，集成 Ollama embedding API，支持文本向量化

问题修复

  • 修复 DateRange 聚合在 Rollup 查询中无法正确合并的问题

优化改进

针对用户使用体验进行了多项改进，包括：

  • 弃用 KNN 模块中的 index.knn 配置项，（此配置项和其他功能经常发生冲突） 简化配置逻辑，该配置项将在后续版本中移除
  • 将 KNN 搜索功能从插件形式集成为内置功能，无需额外安装即可使用
  • 将跨集群复制（CCR）功能从插件形式集成为内置功能，开箱即用
  • 优化索引配置更新验证：增加非动态配置项的值比对，避免误报

INFINI Console v1.29.1

INFINI Console 是一款开源的非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验：

http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

问题修复

  • 修复 agentless 模式下计算索引级别实时 QPS 不准确的问题

INFINI Gateway v1.29.1

INFINI Gateway 是一个开源的面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

优化改进

INFINI Agent v1.29.1

INFINI Agent 负责采集和上传 Elasticsearch, Easysearch, Opensearch 集群的日志和指标信息，通过 INFINI Console 管理，支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。

Agent 本次更新如下：

优化改进

INFINI Loadgen v1.29.1

INFINI Loadgen 是一款开源的专为 Easysearch、Elasticsearch、OpenSearch 设计的轻量级性能测试工具。

Loadgen 本次更新如下：

优化改进

INFINI Framework v1.1.4

INFINI Framework 是 INFINI Labs 基于 Golang 的产品的核心基础，已开源。该框架以开发者为中心设计，简化了构建高性能、可扩展且可靠的应用程序的过程。

Framework 本次更新如下：

功能更新

  • 添加配置选项，以在 WebSocket 连接期间禁用回显消息
  • 允许在 WebSocket 连接/断开时注册回调函数
  • 为 API 添加可选的登录验证配置

优化改进

  • 停止任务后取消任务
  • 回调发生错误时关闭 WebSocket 连接

更多详情请查看以下详细的 Release Notes 或联系我们的技术支持团队！

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

下载地址： https://infinilabs.cn/download

邮件：hello@infini.ltd

电话：(+86) 400-139-9200

Discord：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://infinilabs.cn

INFINI Labs 产品更新 | Console/Gateway/Agent 等产品开源发布首个版本

资讯动态 • INFINI Labs 小助手 发表了文章 • 1 个评论 • 3633 次浏览 • 2024-12-27 16:11 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Easysearch v1.9.0、Gateway、Console、Agent、Loadgen v1.27.0。本次各产品更新了很多亮点功能，如 Easysearch 新增 rollup 功能，优化了多版本兼容配置；Console/Gateway/Agent/Loadgen 及 Framework 开源后，发布首个重大更新版本，支持过期元数据删除，指标图表懒加载，指标采集协程优化等等，欢迎大家下载体验。

INFINI Easysearch v1.9.0

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。详情见：https://infinilabs.cn

Easysearch 本次更新如下：

Improvements

  • 发布 rollup 功能

    • 支持自动对 rollup 索引进行滚动，无需外部触发
    • 支持 avg sum max min value_count percentiles 指标类型的聚合
    • 支持 terms 聚合
    • 支持对指标聚合进行 Pipeline 聚合
    • 支持聚合前先对数据进行过滤
    • 进行聚合查询时支持直接搜索原始索引，不用更改搜索代码
  • 增加适配 logstash 8.x 的请求 header
  • _cat/templates 增加 lifecycle 和 rollover 列的展示

Bug fix

  • 修复 rest-api template 测试错误

INFINI Console v1.27.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。开源地址：https://github.com/infinilabs/console

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Improvements

  • 代码开源，统一采用 Github 仓库进行开发
  • 指标采集优化，由原来的单一协程采集调整为每个注册的集群有单独的协程进行采集
  • 指标监控页面图表展示采用懒加载、单个图表独立加载，增强用户体验
  • 通用时间控件增加超时时间设置
  • 集群选择控件增加注册、刷新功能
  • 提供指标采集状态
  • 表格控件排版优化

Bug fix

  • 修复集群元数据更新不及时问题
  • 修复帮助文档等链接不正确问题
  • 修复节点、索引数据因随机 id 出现重复记录问题
  • 修复 Runtime、Agent 实例编辑页面出错问题
  • 修复集群、节点、索引、分片元数据无 Loading 问题
  • 修复索引健康状态指标采集失败问题
  • 修复个别菜单列未国际化问题

INFINI Gateway v1.27.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。开源地址：https://github.com/infinilabs/gateway

Gateway 本次更新如下：

Improvements

  • 调整队列消费者 slice 默认配置为 1

Bug fix

  • 修复缓存数据丢失导致队列无法消费问题
  • 同步更新 Framework 修复的一些已知问题

INFINI Agent v1.27.0

INFINI Agent 是 INFINI Console 的一个可选探针组件，负责采集和上传集群指标和日志等信息，并可通过 Console 管理。Agent 支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。开源地址：https://github.com/infinilabs/agent

探针 Agent 本次更新如下：

Improvements

  • 与 INFINI Console 统一版本号
  • 同步更新 Framework 修复的已知问题
  • 支持 K8S 环境指标采集

INFINI Loadgen v1.27.0

INFINI Loadgen 是一款轻量、无依赖的 Eaysearch/Elasticsearch/OpenSearch 性能压测工具，支持参数模板化配置，支持压测端均衡流量控制，可以模拟高并发请求。开源地址：https://github.com/infinilabs/loadgen

Loadgen 本次更新如下：

Improvements

  • 保持与 Console 相同版本
  • 同步更新 Framework 修复的已知问题

Bug fix

  • 修复 API 接口测试逻辑异常问题

INFINI Framework

INFINI Framework 是 INFINI Labs 各产品依赖的核心公共代码库。开源地址：https://github.com/infinilabs/framework

Improvements

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

下载地址： https://infinilabs.cn/download

邮件：hello@infini.ltd

电话：(+86) 400-139-9200

Discord：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://infinilabs.cn

INFINI Labs 助力开源与教育：免费许可证计划全面升级

开源项目 • INFINI Labs 小助手 发表了文章 • 0 个评论 • 3348 次浏览 • 2024-06-19 17:21 • 来自相关话题

免费许可证计划

在数字化浪潮席卷全球的今天，INFINI Labs 深刻认识到开源项目和教育机构在技术创新与人才培养中的核心作用。因此，我们郑重推出全新升级的免费许可证计划，旨在全球范围内为开源社区和教育界提供有力支持，共同推动软件生态的繁荣与进步。

一、产品实力与荣誉

1.INFINI Pizza：实时搜索的新纪元

INFINI Pizza

  • 在第十三届“数据技术嘉年华”（DTC2024）上，INFINI Labs 发布了划时代的搜索引擎——INFINI Pizza，标志着搜索型数据库迈入实时搜索的新纪元。
  • INFINI Pizza 凭借先进的设计理念与架构，以及独有的专利技术，实现了对海量数据的无限伸缩，提供高效、准确的实时数据搜索能力

2.行业标杆案例

  • INFINI Labs 荣获中国信通院大数据“星河”标杆案例，其中移动云搜索数据库案例更是荣选为数据库标杆案例。
  • 该案例基于移动云 Easysearch 数据库，通过创新的多集群协同模式，实现了数据高性能存取，展现出极高的经济价值与社会价值。

3.国家发明专利认可

  • INFINI Labs 的多项自主研发技术获得国家发明专利授权，这些成果彰显了公司在大数据领域的技术实力与创新精神。

二、品牌与行业地位

  • INFINI Labs 作为搜索型数据库产品领域的领军企业，积极参与行业标准的制定与推动。

  • 其核心产品 INFINI Easysearch 荣获信通院首批可信搜索型数据库产品证书，再次印证了公司在行业中的领先地位。

三、产品介绍

Products

  1. INFINI Easysearch：作为 Elasticsearch 的国产化替代方案，提供高度兼容性与卓越性能，满足企业级需求。

  2. INFINI Console：轻量级多集群、跨版本搜索基础设施统一管控平台，助力企业高效管理搜索集群。

  3. INFINI Gateway：专为 Elasticsearch 打造的高性能应用网关，提供丰富的功能特性与卓越性能。

  4. INFINI Loadgen：支持多种搜索引擎的轻量级压测工具，为企业提供强大的数据加载与测试能力。

  5. INFINI Pizza：引领实时搜索时代的新星，为企业提供高效、准确的实时数据搜索解决方案。

四、免费许可证计划

1.教育机构学术许可证

  • 面向全球公立或私立学校、职业学校、大学等教育机构，提供非商业用途的软件使用许可。

  • 有效期一年，符合条件的教育机构可继续申请。

2.开源项目许可证

  • 面向非商业开源项目开发者，要求项目拥有活跃社区并在其官网添加 INFINI Labs 的链接。

  • 许可证免费，有效期一年，符合条件的项目可继续申请。

五、申请方式

符合条件的开源项目和教育机构可通过访问 INFINI Labs 官方网站，轻松提交申请，我们将尽快审核并回复。

申请链接：https://infinilabs.cn/community

六、结语

INFINI Labs 以全新升级的免费许可证计划为契机，与全球开源社区和教育界携手合作，共同推动软件生态的创新与发展。让我们共同迎接更加美好的未来！

七、关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.cn

联系我们

原文：https://infinilabs.cn/blog/2024/infini-labs-for-open-source-education-the-free-license-program-upgrades/

极限网关助力好未来 Elasticsearch 容器化升级

Elasticsearch • INFINI Labs 小助手 发表了文章 • 0 个评论 • 4367 次浏览 • 2024-06-12 15:01 • 来自相关话题

极限网关在好未来的最佳实践案例，轻松扛住日增百 TB 数据的流量，助力 ES 从物理机到云原生架构的改造，实现了流控、请求分析、安全管理、无缝迁移等场景。一次完美的客户体验~

背景

物理机架构时代

2022 年，好未来整个日志 Elasticsearch 拥有数十套服务集群，几百台物理机。这么多台机器耗费成本非常高，而且还要花费很大精力去维护。在人力资源有限情况下，存在非常多的弊端，运行成本高，不仅是机器折旧还有机柜等费用。

流量特征

这是来自某个业务线，如下图 1，真实流量，潮汐性非常明显。好未来有很多条业务线，几乎跟这个趋势都一致的，除了个别业务有“续报”、“开课”等活动特殊情况。潮汐性带来的问题就是高峰期 CPU、内存资源是可以消耗很高；低峰期资源使用量非常低，由于是物理架构，这些资源无法给其他业务线共享。

图1

图1

降本增效-容器化改造原动力

日志服务对成本的空前的压力促使我们推进 Elasticsearch 进行架构改造；如何改造，改造成什么样子，这两个问题一直是推进改造原动力。业界能够同时对水平扩展和垂直扩展就是 K8S，我们开始对 Elasticsearch 改造成能在 K8S 上运行进行探索，从而提升 CPU、内存利用率。

物理机时代，没办法把资源动态的扩缩，动态调配，资源隔离，单靠人力操作调度成本太高，几乎无法完成；集群对内存资源需求要比 CPU 资源大很多，由于机器型号配置是固定的，无法“定制”，这也会导致成本居高不下。所以，无论从那个方面来讲，容器化优势非常明显，既能够优化成本，也能够降低运维复杂度。

ES 容器化改造

进行架构升级重点难点- API 服务

改造过程，我们遇到了很多问题，比如容器 ES 版本和物理机 ES 版本不一致，如何让 ES API 能够兼容不同的 ES 版本，由于版本的不兼容，导致无法直接使用原有的 tribenode 进行服务，怎么提供一个高可用的 Elasticsearch API 服务。我们考虑到多个方面，比如使用官方推荐的 proxy 模式、第三方服务等进行选择，经过多方面对比，选择了极限网关 进行 tribenode 替换。

原始 ES API 服务痛点

  • API 访问没有流量控制
  • 可观测性差，而且稳定性一般
  • 版本兼容性差

物理机时代 API 架构

在物理机时代 ES 集群，API 架构如图 2，可以明显看到 tribe node 对所有 ES 集群的“侵入性”是非常大的，这就带来了很多问题，比较严重的就是单个集群对 ES tribenode 的影响和版本升级带来的不兼容问题。

图2

图2

混合时代 API 架构

通过图 3，我们可以看到，极限网关对于版本兼容性很好，能够适配不同的版本。因此，最终选择极限网关作为下一代 ES API 服务方。

图3

图3

里程碑：全部 ES 集群容器化

在 2023 年 3 月份，通过 Elastic 官方 ECK 模式，完成全部日志 ES 集群容器化改造，拥有数百节点，1PB+ 数据存储，每日新增数据 100T 左右。紧接着，除了日志服务外，同时支持了好未来多条业务线。

图4

图4

极限网关实践

下面主要讲述了，为什么选择极限网关，以及极限网关在好未来落地、应用这些内容。

为什么选择极限网关？

学习成本低

我们可以从文档中看到极限网关，其架构简洁，语法简单，直观易懂。学习成本比较低，上手非常快，对新手友好。

性能强悍

经过压测，发现极限网关速度非常快，且针对 Elasticsearch 做了非常细致的优化，能成倍提升写入和查询的速度。

安全性高

支持多种认证方式，最简单的账号密码认证，可以给自定义多个账户密码，大大简化了 Elasticsearch 的安全设置，同时，还可以支持 LDAP 安全验证。

跨版本支持

我们容器化改造过程需要兼容不同版本的 Elasticsrearch，极限网关针对不同的 Elasticsearch 版本做了兼容和针对性处理，能够让业务代码无缝的进行适配，后端 Elasticsearch 集群版本升级能够做到无缝过渡，降低版本升级和数据迁移的复杂度，非常匹配我们的业务场景。

灵活可扩展

可灵活对每个请求进行干预和路由，支持路由的智能学习，内置丰富的过滤器，通过配置动态修改每个请求的处理逻辑，也支持通过插件来进行扩展，满足我们对流量的控制，尤其是限流、用户、IP 等这些功能非常实用。

启用安全策略-为 API 服务保驾护航

痛点

在升级之前使用 tribe 作为 API 服务提供后端，几乎相当于裸奔，没有任何认证策略；另外，tribe 本身的稳定性也有问题，官方在新版本逐渐废弃这种 CCS（跨集群搜索），期间出现多次服务崩溃。

极限网关解决问题

极限网关通过，“basic_auth” 插件，提供最基本的安全校验，使用起来非常方便；同时，极限网关提供 LDAP 插件，可以接入公共的 LDAP 服务，对所有的访问用户进行校验，安全策略对所有的用户生效，不用担心因为 IP 问题泄漏数据等。

强大的过滤功能

在使用 ES 集群过程中，许多场景，需要对请求进行控制、限制等操作。在这方便，感受到了极限网关强大的产品力。比如下面的两个场景

对异常流量进行限流

  • 支持对 IP 限流
  • 支持对 hostname 限流
  • 支持 header 限流

对异常用户进行封禁

当 Elasticsearch 是通过 Basic Auth 或者 LDAP 方式来进行身份认证的时候，request_user_filter 过滤器可用来按请求的用户名信息来进行过滤。操作起来也非常简单，只需要 request_user_filter 这一个过滤器。

- request_user_filter:
    include:
        - "elastic"
    exclude：
        - "Ryan"

总结来讲，主要有这些方面的功能：

图5

图5

优秀的可观测性

痛点

改造前经常为看不到直观的数据指标感到头疼，查看指标需要多个地方同时打开，去筛选，查找，非常繁琐，付出的成本非常大。为此，大家都再考虑如何优化这种情况，无奈优先级比较低，一直没有真正的投入时间去优化这块。

完美解决

使用了极限网关，通过收集请求日志，非常清晰的收集到想要的数据，具体如下：

  • 总体方面：
    • 流量曲线
    • 状态码占比
    • 缓存统计
    • 每台网关请求流量
  • 细节方面：
    • 打印每次请求语句
    • 可以查看请求到具体 ES 节点流量
    • 可以查看过滤器的列表

通过下图，我们可以从管理视角直观的看到各种信息，这对于管理员来讲，省时省力，方便快捷。

图6

图6

意外收获：无缝迁移业务 Elasticsearch 上云

由于前期日志业务上云，受到非常好的反馈，多个业务线期望能够上云上服务，达到降本增效的目的。

支持双写

数据可以通过极限网关同时写入两个 ES 集群，能够保障数据完全一致，安全可靠。

无缝切换

切换很丝滑，影响非常小，能够让外界几乎感受不到服务波动。

图7

图7

通过使用极限网关，自建 ES 集群可以无缝的迁移上云，在整个迁移的过程中，两套集群通过网关进行了解耦，在迁移的过程中还能实现版本的无缝升级，极大降低了迁移成本，提高迁移效率，多次验证服务稳定可靠。

极限网关流量概览

这是其中一套极限网关的流量统计。用这部分数据进行巡检，一目了然，做到全局的掌控，提高感知力度。

图8

图8

极限网关使用总结

极限网关提供一系列高性能和高可靠性的网关服务。使用这样的服务给我们带来以下好处：

  1. 可观测性好：极限网关可以动态的对 Elasticsearch 运行过程中请求进行拦截和分析，通过指标和日志来了解集群运行状态，这些指标可以用于提升性能和业务优化。
  2. 增强安全性：包含先进的安全机制，如 basicauth、LDAP 等支持，保护用户数据不受未授权访问和各种网络威胁的侵害。
  3. 高稳定性：通过冗余设计和故障转移机制，极限网关能够确保网络服务的高可用性，即使在某些组件发生故障时也能保持服务不中断，单版本最长服务超过 15 个月。
  4. 易于管理：通过提供 INFINI Console 简洁直观的管理界面，让用户能够轻松配置和监控网络状态，提升管理效率。
  5. 客户支持：良好的客户服务支持可以帮助用户快速解决使用过程中遇到的问题，提供专业的技术指导。

综上所述，极限网关为用户提供了一个高速、安全、稳定且易于管理的 ES 网关，适合对网络性能有较高要求的个人和企业用户。

未来规划

第一阶段，完成了日志 ES 集群，所有集群的容器化改造，合并，成功的把成本降低了 60%以上。这期间积累了丰富容器化经验，为业务 ES 集群上容器做了良好的铺垫；成本优势和运维优势吸引越来越多的业务接入到容器化 ES 集群。

提升 ES 集群效能--新技术应用&&版本升级

  • 极限科技官方推荐的 Easysearch 在压缩率，查询速度等等方面有很多的优势，通过长时间的测试稳定性，新特性，对比云原生的 ES 集群，根据测试结果，给“客户”提供多种选择，这也是工作重点之一。
  • 我们当前使用的 ES 版本是 6.8，已经远远落后于官方版本，今年我们计划在选择合适的集群升级 ES 版本，拥抱更多官方提供的特性。

混合（多）云架构支持

随着越来越多的 ES 集群在机房的 K8S 集群部署，这里资源出现了紧张局面。 我们尝试在云上部署自建 ES 集群，弥补机房资源有限，无法大规模扩容，同时能够支持多活场景，满足更多客户的不同需求。混合云主要实现以下几种能力：

1、扩缩容：满足不同业务灵活适配

混合（多）云部署，可以让负载内部私有云 ES，同时部署到公有云，提升扩展 IT 基础设施不仅局限于 CPU、内存，还有存储。比如某一个业务要做活动，预估流量“大爆发”，需要提前准备大规模资源，在机房内根本来不及采购扩容支持，然而在公有云上就能很方便扩容、缩容。在云上搭建 ES 集群，设置满足需求的数量、容量、配置，配合极限网关路由策略，精准的把控流量流向。

图9

图9

2、灾备：紧急情况快速部署，恢复 ES 集群读写

当机房级别大规模故障，部分业务实现了多活，单一的机房故障不会影响其服务能力，而此时比如日志查看等仍有需求，为了满足这部分“客户”需求，可以在云上 K8S 集群，快速搭建 ES 集群，恢复日志读写功能。

图10

图10

参考文档：

作者：张华勋，前新浪 CDN 研发，工作主要涉及 Mysql、MongoDB、Redis、Elasticsearch、流量调度等组件和系统，以及运维自动化、平台化等工作。现就职于好未来。

关于好未来

好未来（NYSE：TAL）是一家以内容能力与科技能力为基础，以科教、科创、科普为战略方向，助力人的终身成长，并持续探索创新的科技公司。 好未来的前身学而思成立于 2003 年，2010 年在美国纽交所正式挂牌交易。好未来以“爱与科技助力终身成长”为使命，致力成为持续创新的组织。更多参见：https://www.100tal.com/

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.cn

联系我们

INFINI Labs 产品更新 | Easysearch 1.8.2 发布优化 CCR 性能

Easysearch • liaosy 发表了文章 • 0 个评论 • 3243 次浏览 • 2024-06-09 12:52 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Easysearch v1.8.2、Gateway、Console、Agent、Loadgen v1.26.0。本次各产品更新了很多亮点功能，如 Easysearch 优化 CCR 同步性能；Gateway 增加了 HTTP 请求动态域名路由功能，移除了安全相关的 Filter，进一步提升 Gateway 稳定性；Console 修复了多个已知问题，如当文档数过亿时单位换算错误，修复了因采集延迟导致指标图表显示异常，修复了多行查询中包含 SQL 查询异常等问题。欢迎大家下载体验。

以下是本次更新的详细说明。

INFINI Easysearch v1.8.2

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Bug fix

  • 修复 source_reuse 与 object 字段为 enable: false 时的冲突

Improvements

  • 升级部分依赖包版本，Commons-collections to 3.2.2, Snakeyaml to 2.0
  • 优化 CCR 同步性能及调整 CCR 全局配置参数
  • 优化插件配置命名，去除"plugins."
  • 优化配置文件目录获取命名

INFINI Console v1.26.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Bug fix

  • 修复监控数据布局
  • 修复命令存储权限
  • 修复多行请求包含 SQL 语法
  • 修复文档数过亿时换算错误
  • 修复导入低版本 v1.6.0 告警规则缺少字段问题
  • 修复当 buck_size 小于 60 秒时，因指标采集延迟导致指标显示异常问题

INFINI Gateway v1.26.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Improvements

  • feat: add wildcard_domain filter
  • chore: remove security filter and translog_viewer

INFINI Framework

INFINI Framework 是 INFINI Labs 各产品依赖的内部核心公共代码库。

Framework 本次更新如下：

Improvements

  • feat: support dynamic app setting
  • feat: add cluster settings query args
  • feat: add gateway config
  • feat: add http interceptor
  • feat: return host info in info api
  • feat: add util to convert string to float
  • feat: use common app setting api to instead of auth setting api
  • feat: crontab task support multi crontab expression
  • fix: skip submit empty bulk requests
  • feat: support ccr api
  • fix: get latest offset should compare segment first
  • fix: wrong use of zstd with vfs
  • fix: prevent close closed channel
  • fix: panic on error while saving keystore

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.cn

INFINI Labs 产品更新 | Console 1.24.0 操作日志审计功能发布

资讯动态 • liaosy 发表了文章 • 0 个评论 • 3903 次浏览 • 2024-04-28 12:51 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Console，Gateway 1.24.0。本次各产品更新了很多亮点功能，如 Console 增加操作日志审计功能，优化数据探索字段统计，修复 Gateway 增加认证后添加实例失败等问题。以下是本次更新的详细说明。

INFINI Console v1.24.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 用户操作审计日志功能
  • 新增告警规则克隆一键克隆功能，简化重复类型告警的创建

Bug fix

  • 修复普通用户权限 403 问题
  • 修复探针管理表格展开显示错位问题

Improvements

  • 优化开发工具集群选择控件显示位置
  • 优化数据探索查询控件显示宽度
  • 优化数据探索字段统计功能
  • 优化告警规则列表页搜索，支持远程搜索
  • Discover 左侧字段聚合支持开关控制拉取本地或远程统计值

INFINI Gateway v1.24.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Improvements

  • Refactoring http client tls config
  • Write field routing to bulk metadata when field _routing exists in scrolled doc

Bug fix

  • Fix(reshuffle filter): make sure queue config always have label type
  • Fix rotate config usage

INFINI Framework

Improvements

  • feat: allow to use default auth for agent’s auto enroll
  • refactor: refactor func GetFieldCaps
  • feat: register background job to clean up badger LSM tree
  • fix: skip load missing wal

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.cn

INFINI Labs 产品更新 | Easysearch 1.7.1发布，改进跨集群复制的数据加载等

Easysearch • liaosy 发表了文章 • 0 个评论 • 2691 次浏览 • 2024-03-04 15:25 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Console，Gateway，Agent 1.23.0 和 Easysearch 1.7.1。本次各产品更新了很多亮点功能，如 Console 优化实例管理中增加磁盘空闲空间显示，Easysearch 改进 HierarchyCircuitBreakerService 并添加断路器、改进跨集群复制的数据加载，增加对 source_reuse 索引的支持等。欢迎大家下载体验。

以下是本次更新的详细说明。

INFINI Easysearch v1.7.1

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Bug fix

  • 修复 source_reuse 与字段别名冲突
  • 改进 HierarchyCircuitBreakerService 并添加断路器
  • 修复 _meta 不为空且 启用 source_reuse 时的映射解析错误
  • 修复 source_reuse 下对多值还原不正确的问题

Improvements

  • 改进 HierarchyCircuitBreakerService 并添加断路器
  • 改进跨集群复制的数据加载，增加对 source_reuse 索引的支持

INFINI Console v1.23.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Bug fix

  • 修复数据迁移中数据分片范围因精度导致数据溢出显示为负数
  • 修复删除实例队列后消费的 Offset 未重置问题
  • 修复网友提出的各种问题，如集群设置默认打开节点、索引采集等

Improvements

  • 优化初始化配置向导，分步骤执行
  • 优化实例管理中增加磁盘空闲空间显示
  • 优化实例队列名称显示

INFINI Gateway v1.23.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Bug fix

  • 修复删除实例队列后消费的 Offset 未重置问题

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

如何防止 Elasticsearch 服务 OOM？

Easysearch • yangmf2040 发表了文章 • 0 个评论 • 3115 次浏览 • 2024-02-26 10:12 • 来自相关话题

Elasticsearch（简称：ES） 和传统关系型数据库有很多区别， 比如传统数据中普遍都有一个叫“最大连接数”的设置。目的是使数据库系统工作在可控的负载下，避免出现负载过高，资源耗尽，谁也无法登录的局面。

那 ES 在这方面有类似参数吗？答案是没有，这也是为何 ES 会被流量打爆的原因之一。

针对大并发访问 ES 服务，造成 ES 节点 OOM，服务中断的情况，极限科技旗下的 INFINI Gateway 产品（以下简称 “极限网关”）可从两个方面入手，保障 ES 服务的可用性。

  1. 限制最大并发访问连接数。
  2. 限制非重要索引的请求速度，保障重要业务索引的访问速度。

下面我们来详细聊聊。

架构图

所有访问 ES 的请求都发给网关，可部署多个网关。

限制最大连接数

在网关配置文件中，默认有最大并发连接数限制，默认最大 10000。

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 10000
    network:
      binding: $[[env.GW_BINDING]]
      # See `gateway.disable_reuse_port_by_default` for more information.
      reuse_port: true

使用压测程序测试，看看到达10000个连接后，能否限制新的连接。 超过的连接请求，被丢弃。更多信息参考官方文档。

限制索引写入速度

我们先看看不做限制的时候，测试环境的写入速度，在 9w - 15w docs/s 之间波动。虽然峰值很高，但不稳定。 接下来，我们通过网关把写入速度控制在最大 1w docs/s 。 对网关的配置文件 gateway.yml ，做以下修改。

env: # env 下添加
    THROTTLE_BULK_INDEXING_MAX_BYTES: 40485760 #40MB/s
    THROTTLE_BULK_INDEXING_MAX_REQUESTS: 10000 #10k docs/s
    THROTTLE_BULK_INDEXING_ACTION: retry #retry,drop
    THROTTLE_BULK_INDEXING_MAX_RETRY_TIMES: 10 #1000
    THROTTLE_BULK_INDEXING_RETRY_DELAY_IN_MS: 100 #10

router: # route 部分修改 flow
  - name: my_router
    default_flow: default_flow
    tracing_flow: logging_flow
    rules:
      - method:
          - "*"
        pattern:
          - "/_bulk"
          - "/{any_index}/_bulk"
        flow:
          - write_flow

flow: #flow 部分增加下面两段    
  - name: write_flow
    filter:
      - flow:
          flows:
            - bulking_indexing_limit
      - elasticsearch:
          elasticsearch: prod
          max_connection_per_node: 1000
  - name: bulking_indexing_limit
    filter:
      - bulk_request_throttle:
          indices:
            "test-index":
              max_bytes: $[[env.THROTTLE_BULK_INDEXING_MAX_BYTES]]
              max_requests: $[[env.THROTTLE_BULK_INDEXING_MAX_REQUESTS]]
              action: $[[env.THROTTLE_BULK_INDEXING_ACTION]]
              retry_delay_in_ms: $[[env.THROTTLE_BULK_INDEXING_RETRY_DELAY_IN_MS]]
              max_retry_times: $[[env.THROTTLE_BULK_INDEXING_MAX_RETRY_TIMES]]
              message: "bulk writing too fast" #触发限流告警message自定义
              log_warn_message: true

再次压测，test-index 索引写入速度被限制在了 1w docs/s 。

限制多个索引写入速度

上面的配置是针对 test-index 索引的写入速度控制。如果想添加其他的索引，新增一段配置即可。 比如，我允许 abc 索引写入达到 2w docs/s，test-index 索引最多不超过 1w docs/s ，可配置如下。

  - name: bulking_indexing_limit
    filter:
      - bulk_request_throttle:
          indices:
            "abc":
              max_requests: 20000
              action: drop
              message: "abc doc写入超阈值" #触发限流告警message自定义
              log_warn_message: true
            "test-index":
              max_bytes: $[[env.THROTTLE_BULK_INDEXING_MAX_BYTES]]
              max_requests: $[[env.THROTTLE_BULK_INDEXING_MAX_REQUESTS]]
              action: $[[env.THROTTLE_BULK_INDEXING_ACTION]]
              retry_delay_in_ms: $[[env.THROTTLE_BULK_INDEXING_RETRY_DELAY_IN_MS]]
              max_retry_times: $[[env.THROTTLE_BULK_INDEXING_MAX_RETRY_TIMES]]
              message: "bulk writing too fast" #触发限流告警message自定义
              log_warn_message: true

限速效果如下

限制读请求速度

我们先看看不做限制的时候，测试环境的读取速度，7w qps 。 接下来我们通过网关把读取速度控制在最大 1w qps 。 继续对网关的配置文件 gateway.yml 做以下修改。

  - name: default_flow
    filter:
      - request_path_limiter:
          message: "Hey, You just reached our request limit!"                                      rules:      
            - pattern: "/(?P<index_name>abc)/_search"                          
              max_qps: 10000
              group: index_name                                    
      - elasticsearch: 
          elasticsearch: prod                            
          max_connection_per_node: 1000

再次进行测试，读取速度被限制在了 1w qps 。

限制多个索引读取速度

上面的配置是针对 abc 索引的写入速度控制。如果想添加其他的索引，新增一段配置即可。 比如，我允许 abc 索引读取达到 1w qps，test-index 索引最多不超过 2w qps ，可配置如下。

  - name: default_flow
    filter:
      - request_path_limiter:
          message: "Hey, You just reached our request limit!"
          rules:
            - pattern: "/(?P<index_name>abc)/_search"
              max_qps: 10000
              group: index_name
            - pattern: "/(?P<index_name>test-index)/_search"
              max_qps: 20000
              group: index_name
      - elasticsearch:
          elasticsearch: prod
          max_connection_per_node: 1000

多个网关限速

限速是每个网关自身的控制，如果有多个网关，那么后端 ES 集群收到的请求数等于多个网关限速的总和。 本次介绍就到这里了。相信大家在使用 ES 的过程中也遇到过各种各样的问题。欢迎大家来我们这个平台分享自己的问题、解决方案等。如有任何问题，请随时联系我，期待与您交流！

INFINI Labs 产品更新 | Console 数据迁移支持 Percentiles 均匀分区

资讯动态 • liaosy 发表了文章 • 0 个评论 • 3807 次浏览 • 2023-12-30 16:25 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Console v1.14.0，Gateway 1.21.0。其中 Console 数据迁移支持 Percentiles 均匀分区，修复已知 Bug 等。以下是本次更新的详细说明。

INFINI Console v1.14.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 数据迁移支持 decimal 类型分区设置
  • 数据迁移支持 number 类型均匀分区

Bug fix

  • 修复告警恢复后，新周期内没有发送告警通知消息的问题

INFINI Gateway v1.21.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Bug fix

  • fix log error when temp file was missing

INFINI Framework

INFINI Framework 是 INFINI Labs 各产品依赖的内部核心公共代码库。

Framework 本次更新如下：

  • feat: support even partition
  • feat: support nested object
  • fix log error when temp file was missing

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

INFINI Labs 产品更新 | 修复 Easysearch 跨集群复制索引同步问题，Gateway 内存异常增长等问题

Easysearch • liaosy 发表了文章 • 0 个评论 • 3021 次浏览 • 2023-12-01 18:45 • 来自相关话题

release

INFINI Labs 产品又更新啦~，本次更新主要对 Easysearch、Gateway、Console、Agent 等产品功能进行优化和相关 Bug 修复，解决了内存异常增长等问题，以下是详细说明。

INFINI Easysearch v1.6.2

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Bug fix

  • 修复跨集群复制（CCR）不能对自动滚动生成的索引进行同步的问题

Improvements

  • 优化初始化脚本，增加-s/-slient 参数，自动安装。
  • 新增含 jdk/plugins 的 bundle 安装包

INFINI Gateway v1.20.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Bug fix

  • 修复由 Framework Bug 造成连接数不释放、内存异常增长的问题

Improvements

  • 增加配置，允许设置 fasthttp client 相关参数

INFINI Console v1.12.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Bug fix

  • 修复数据探索 multi fields 字段计算 top values 报错的问题
  • 修复由 Framework Bug 造成连接数不释放、内存异常增长的问题
  • 修复内网模式下静态资源远程加载的问题
  • 修复数据看板数据源配置校验异常的问题

Improvements

  • 优化数据探索计算 top values，使用先采样后，后取 top values
  • 可通过配置参数 http_client.read_buffer_size 设置读取缓存大小，解决开发工具执行命令时，默认缓存太小的问题

INFINI Agent v0.7.1

INFINI Agent 是 INFINI Console 的一个可选探针组件，负责采集和上传集群指标和日志等信息，并可通过 Console 管理。Agent 支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。

Agent 本次更新如下：

Features

  • 添加 http processor

Bug fix

  • 修复由 Framework Bug 造成连接数不释放、内存异常增长的问题

Improvements

  • 进一步优化内存占用，降到 50M 以下

INFINI Framework

INFINI Framework 是 INFINI Labs 各产品依赖的内部核心公共代码库。

Framework 本次更新如下：

  • fix: fix the issue of disk queue was blocked
  • chore: checkout specify branch before pull

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

INFINI Labs 产品更新 | 发布 Easysearch Java 客户端，Console 支持 SQL 查询等功能

Easysearch • liaosy 发表了文章 • 0 个评论 • 3711 次浏览 • 2023-11-17 18:56 • 来自相关话题

release

INFINI Labs 产品又更新啦~，本次更新概要如下：发布 Easysearch-client Java 客户端，开发者通过 client 与 Easysearch 集群的交互变得更加简洁和直观；Console 开发工具新增 SQL 特性，支持 SELECT 查询等语法高亮和自动提示等；Gateway 的系统 API 添加了基于基本身份验证的安全功能。

以下是本次更新的详细说明。

INFINI Easysearch-client v1.0.1

正式发布 Easysearch Java 客户端。

这一里程碑式的更新为开发人员带来了前所未有的便利性，使得与 Easysearch 集群的交互变得更加简洁和直观。现在，通过 Easysearch-client 客户端，开发者可以直接使用 Java 方法和数据结构来进行交互，而不再需要依赖于传统的 HTTP 方法和 JSON。这一变化大大简化了操作流程，使得数据管理和索引更加高效。高级客户端的功能范围包括处理数据操作，管理集群，包括查看和维护集群的健康状态，并对 Security 模块全面兼容。它提供了一系列 API，用于管理角色、用户、权限、角色映射和账户。这意味着安全性和访问控制现在可以更加细粒度地管理，确保了数据的安全性和合规性。

使用说明参见：快速开始

INFINI Console v1.11.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 开发工具 SQL 查询支持
    • 支持 SELECT 查询及语法高亮
    • 支持索引和字段自动提示
    • 支持 FROM 前置语法

Bug fix

  • 修复平台概览集群指标为空的问题

Improvements

  • LDAP 支持从 DN 中解析 OU 属性
  • 集群动态优化显示，新增节点名称和索引名称的聚合统计过滤

INFINI Gateway v1.19.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Features

  • 添加 http 处理器
  • 在 API 模块中添加基于基本身份验证的安全性
  • 允许将自身注册到配置管理器
  • 允许在配置错误时触发 panic

Bug fix

  • 修复 rewrite_to_bulk 在较新版本中缺少 _type 的问题
  • 修复 rewrite_to_bulk，支持无索引文档操作

Improvements

  • 更新测试，断言解析结果

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

INFINI Labs 产品更新 | Gateway 支持基于 Kafka 的复制能力，发布 Helm Charts 部署方式

资讯动态 • liaosy 发表了文章 • 0 个评论 • 2858 次浏览 • 2023-09-22 16:16 • 来自相关话题

release

INFINI Labs 产品又更新啦~。本次更新概要如下：Easysearch 新增了索引字段相关统计 API，优化了 source_reuse 提升压缩效率；Gateway 新增诸多新特性，如：支持基于 Kafka 的复制能力，添加可插拔的分布式锁实现，新增 CPU 资源限制等功能；Console 本次主要优化了数据迁移功能，迁移任务详情页新增了若干指标图和日志查看等功能。

欢迎大家下载使用和反馈。

INFINI Helm Charts v0.1.0

INFINI Helm Charts 是一组 Kubernetes 部署包管理工具。基于 Helm Charts，我们将 INFINI Labs 旗下相关产品预先配置好程序资源包，大大简化了部署流程。Github 仓库地址：https://github.com/infinilabs/helm-charts。

Helm Charts 本次更新如下：

Features

  • 添加 Console Chart
  • 添加 Easysearch Chart，支持单节点以及多节点（节点角色可配置）部署

部署视频演示：

博客文章：

INFINI Easysearch v1.6.0

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Features

  • 新增 _field_usage_stats API，统计索引每个字段的访问次数
  • 新增 _disk_usage API，可以分析指定索引每个字段的磁盘占用大小
  • 增加 flattened 类型，将 JSON 对象作为字符串处理，可以减少嵌套 JSON 型的文档的大小

Improvements

  • source_reuse 增加对 _source 中数字类型的值进行复用压缩，可进一步降低 _source 磁盘占用
  • 改进 source_reuse 筛选字段的逻辑

INFINI Gateway v1.18.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Breaking changes

  • 彻底移除了 request_body_truncate 和 response_body_truncate 过滤器

Features

  • 支持基于 Kafka 的复制能力
  • 在请求上下文中添加 _util.generate_uuid
  • 在请求上下文中添加 _util.increment_id.BUCKET_NAME
  • 在 Pipeline 配置中添加 singleton，防止多个 Pipeline 同时运行
  • 添加可插拔的分布式锁实现
  • 添加通用应用程序的 preference 配置
  • 泛化队列抽象，重构磁盘队列，完善 Kafka 实现
  • 添加 merge_to_bulk 处理器， 废弃 indexing_merge 处理器
  • 添加 flow_replay 处理器，废弃 flow_runner 处理器
  • 为复制场景添加 replication_correlation
  • 添加 hash_mod 过滤器
  • 在 bulk_response_process 过滤器中添加新参数
  • 添加 request_reshuffle 过滤器
  • 添加资源限制，允许设置最大 CPU 数或绑定亲和性
  • 支持模板中的嵌套变量
  • 添加 rewrite_to_bulk 过滤器

Bug fix

  • 修复了 Pipeline 中重试延迟未生效的问题
  • 修复了模板中不支持数字的问题
  • 修复了队列选择器通过标签的问题，如果指定了多个标签，它们都应该一起匹配

Improvements

  • 将所有模块名称转换为小写
  • 在启动期间预取 Elasticsearch 元数据
  • 添加应用程序范围的关闭信号
  • 重构队列 API，支持 Kafka 管理
  • 在 Badger 模块中添加 enabled
  • 允许使用优先级注册模块/插件
  • 统一队列的使用和初始化
  • 优化 bulk_reshuffle 过滤器的性能，添加响应头 X-Bulk-Reshuffled
  • 支持在 queue 过滤器中使用变量，允许输出最后生成的消息偏移量

INFINI Console v1.8.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 数据迁移任务支持自定义名称和添加标签
  • 数据迁移任务详情页新增若干指标
  • 数据迁移任务详情页新增查看日志

Improvements

  • 数据迁移 UI 优化
  • 优化监控报表、数据看板、数据探索的时间控件 UI

数据迁移列表页

数据迁移详情页

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.com/invite/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

INFINI Labs 产品更新 | Easysearch 新增分词插件、Gateway 支持邮件发送等功能

资讯动态 • liaosy 发表了文章 • 0 个评论 • 2567 次浏览 • 2023-07-21 17:28 • 来自相关话题

INFINI Labs 产品又更新啦~，本次更新概要如下：Easysearch 新增了分词插件、优化了生命周期管理功能等；Gateway 新增 smtp 过滤器来支持邮件的发送，支持自动跳过因为异常关闭而损坏的磁盘队列文件等；Console 新增熔断器监控指标、新增矩形树图(Treemap)、优化了探针 Agent 指标采集和集群自动关联操作等。欢迎大家下载体验。

INFINI Easysearch v1.4.0

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 衍生自基于开源协议 Apache 2.0 的 Elasticsearch 7.10 版本。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Features

  • 索引生命周期管理增加 wait_for_snapshot 操作，在删除索引之前，等待执行指定的快照管理策略，这样可以确保已删除索引的快照可用
  • 增加 analysis-hanlp 分词插件
  • 增加 jieba 分词插件

Bug fix

  • 修复启用 index.source_reuse 时，对复杂多层 json 的 source 字段 解析不正确的 Bug

Improvements

  • 更新索引生命周期管理 api 文档，增加策略应用和更新说明，增加 wait_for_snapshot 说明
  • 执行 initialize.sh 命令时增加初始化确认提示，是否将 admin 密码记录日志

INFINI Gateway v1.17.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Features

  • 新增 consumer Processor 来标准化订阅消息队列
  • 新增 smtp 过滤器来支持邮件的发送

Bug fix

  • 支持自动跳过因为异常关闭而损坏的磁盘队列文件

INFINI Console v1.4.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 新增熔断器监控指标
  • 网关队列管理支持多选删除消费者
  • 数据看板新增组件矩形树图

Bug fix

  • 修复开发工具智能提示兼容性问题
  • 修复探针列表状态显示异常的问题
  • 修复探针列表分页不生效的问题
  • 修复数据看板 统计函数显示不对的问题
  • 修复探针下发采集指标配置重复的问题
  • 修复数据看板设置过滤条件不生效的问题
  • 修复主机列表探针状态不对的问题
  • 修复网关管理删除队列不成功的问题
  • 修复数据探索当前集群没索引时跳转的问题
  • 修复数据看板编辑状态下点击事件的问题

Improvements

  • 探针进程关联支持通过选择集群自动关联，简化操作
  • 探针列表支持排序
  • 探针支持向上滚动查看节点日志
  • 采集监控指标添加 cluster_uuid 信息
  • 数据看板支持配置指标排序

INFINI Agent v0.6.0

INFINI Agent 是 INFINI Console 的一个可选探针组件，负责采集和上传集群指标和日志等信息，并可通过 Console 管理。Agent 支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。

Agent 本次更新如下：

Features

  • 采集监控指标添加 cluster_uuid 信息

Improvements

  • 修复发现节点进程信息时获取不到最新集群配置的问题

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

或者通过 Discord 渠道加入聊天室：https://discord.com/invite/tnZ8S5vQ

欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群讨论，或者扫码加入我们的知识星球一起学习交流。

联系我们

关于极限科技（INFINI Labs）

关于极限科技

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

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

官网：https://www.infinilabs.com

用极限网关实现ES容灾，简单！

Elasticsearch • yangmf2040 发表了文章 • 0 个评论 • 3508 次浏览 • 2023-07-20 10:33 • 来自相关话题

身为 IT 人士，大伙身边的各种系统肯定不少吧。系统虽多，但最最最重要的那套、那几套，大伙肯定是捧在手心，关怀备至。如此重要的系统，万一发生故障了且短期无法恢复，该如何保障业务持续运行？ 有过这方面思考或经验的同学，肯定脱口而出--切灾备啊。 是的，接下来我来介绍下我们的 ES 灾备方案。当然如果你有更好的，请使用各种可用的渠道联系我们。

总体设计

通过极限网关将应用对主集群的写操作，复制到灾备集群。应用发送的读请求则直接转发到主集群，并将响应结果转发给应用。应用对网关无感知，访问方式与访问 ES 集群一样。

方案优势

  • 轻量级

极限网关使用 Golang 编写，安装包很小，只有 10MB 左右，没有任何外部环境依赖，部署安装都非常简单，只需要下载对应平台的二进制可执行文件，启动网关程序的二进制程序文件执行即可。

  • 跨版本支持

极限网关针对不同的 Elasticsearch 版本做了兼容和针对性处理，能够让业务代码无缝的进行适配，后端 Elasticsearch 集群版本升级能够做到无缝过渡，降低版本升级和数据迁移的复杂度。

  • 高可用

极限网关内置多种高可用解决方案，前端请求入口支持基于虚拟 IP 的双机热备，后端集群支持集群拓扑的自动感知，节点上下线能自动发现，自动处理后端故障，自动进行请求的重试和迁移。

  • 灵活性

主备集群都是可读可写，切换迅速，只需切换网关到另一套配置即可。回切灵活，恢复使用原配置即可。

架构图

网关程序部署

下载

根据操作系统和平台选择下面相应的安装包： 解压到指定目录：

mkdir gateway
tar -zxf xxx.gz -C gateway

修改网关配置

在此 下载 网关配置，默认网关会加载配置文件 gateway.yml ，如果要指定其他配置文件使用 -config 选项指定。 网关配置文件内容较多，下面展示必要部分。

  #primary
  PRIMARY_ENDPOINT: http://192.168.56.3:7171
  PRIMARY_USERNAME: elastic
  PRIMARY_PASSWORD: password
  PRIMARY_MAX_QPS_PER_NODE: 10000
  PRIMARY_MAX_BYTES_PER_NODE: 104857600 #100MB/s
  PRIMARY_MAX_CONNECTION_PER_NODE: 200
  PRIMARY_DISCOVERY_ENABLED: false
  PRIMARY_DISCOVERY_REFRESH_ENABLED: false
  #backup
  BACKUP_ENDPOINT: http://192.168.56.3:9200
  BACKUP_USERNAME: admin
  BACKUP_PASSWORD: admin
  BACKUP_MAX_QPS_PER_NODE: 10000
  BACKUP_MAX_BYTES_PER_NODE: 104857600 #100MB/s
  BACKUP_MAX_CONNECTION_PER_NODE: 200
  BACKUP_DISCOVERY_ENABLED: false
  BACKUP_DISCOVERY_REFRESH_ENABLED: false

PRIMARY_ENDPOINT：配置主集群地址和端口
PRIMARY_USERNAME、PRIMARY_PASSWORD: 访问主集群的用户信息
BACKUP_ENDPOINT：配置备集群地址和端口
BACKUP_USERNAME、BACKUP_PASSWORD: 访问备集群的用户信息

运行网关

前台运行 直接运行网关程序即可启动极限网关了，如下：

./gateway-linux-amd64

后台运行

./gateway-linux-amd64 -service install
Success
./gateway-linux-amd64 -service start
Success

卸载服务

./gateway-linux-amd64 -service stop
Success
./gateway-linux-amd64 -service uninstall
Success

灾备功能测试

在灾备场景下，为保证数据一致性，对集群的访问操作都通过网关进行。注意只有 bulk API 的操作才会被复制到备集群。 在此次测试中，网关灾备配置功能为：

  • 主备集群正常时

读写请求正常执行； 写请求被记录到队列，备集群实时消费队列数据。

  • 当主集群故障时

写入请求报错，主备集群都不写入数据； 查询请求转到备集群执行，并返回结果给客户端。

  • 当备集群故障时

读写请求都正常执行； 写操作记录到磁盘队列，待备集群恢复后，自动消费队列数据直到两个集群一致。

主备集群正常时写入、查询测试

写入数据

# 通过网关写入数据
curl -X POST "localhost:18000/_bulk?pretty" -H 'Content-Type: application/json' -uelastic:password -d'
{ "index" : { "_index" : "test", "_id" : "1" } }
{ "field1" : "value1" }
{ "create" : { "_index" : "test", "_id" : "2" } }
{ "field2" : "value2" }
'

查询数据

# 查询主集群
curl 192.168.56.3:7171/test/_search?pretty -uelastic:password

# 查询备集群
curl 192.168.56.3:9200/test/_search?pretty -uadmin:admin

# 查询网关，网关转发给主集群执行
curl 192.168.56.3:18000/test/_search?pretty -uelastic:password

主备集群都已写入数据，且数据一致。通过网关查询，也正常返回。

删除和更新文档

# 通过网关删除和更新文档
curl -X POST "192.168.56.3:18000/_bulk?pretty" -H 'Content-Type: application/json' -uelastic:password -d'
{ "delete" : { "_index" : "test", "_id" : "1" } }
{ "update" : {"_id" : "2", "_index" : "test"} }
{ "doc" : {"field2" : "value2-updated"} }
'

查询数据
# 查询主集群
curl 192.168.56.3:7171/test/_search?pretty -uelastic:password

# 查询备集群
curl 192.168.56.3:9200/test/_search?pretty -uadmin:admin

两个集群都已执行删除和更新操作，数据一致。

主集群故障时写入、查询测试

为模拟主集群故障，直接关闭主集群。

写入数据

# 通过网关写入数据
curl -X POST "192.168.56.3:18000/_bulk?pretty" -H 'Content-Type: application/json' -uelastic:password -d'
{ "index" : { "_index" : "test", "_id" : "3" } }
{ "field3" : "value3" }
{ "create" : { "_index" : "test", "_id" : "4" } }
{ "field4" : "value4" }
'

写入数据报错

查询数据

# 通过网关查询，因为主集群不可用，网关将查询转发到备集群执行
curl 192.168.56.3:18000/test/_search?pretty -uelastic:password

正常查询到数据，说明请求被转发到了备集群执行。

备集群故障时写入、查询测试

为模拟备集群故障，直接关闭备集群。

写入数据

# 通过网关写入数据
curl -X POST "192.168.56.3:18000/_bulk?pretty" -H 'Content-Type: application/json' -uelastic:password -d'
{ "index" : { "_index" : "test", "_id" : "5" } }
{ "field5" : "value5" }
{ "create" : { "_index" : "test", "_id" : "6" } }
{ "field6" : "value6" }
'

数据正常写入。

查询数据

# 通过网关查询
curl 192.168.56.3:18000/test/_search?pretty -uelastic:password

查询成功返回。主集群成功写入了两条新数据。同时此数据会被记录到备集群的队列中，待备集群恢复后，会消费此队列追数据。

恢复备集群

启动备集群。

查询数据

等待片刻或通过 INFINI Console 确定网关队列消费完毕后，查询备集群的数据。 (生产和消费 offset 相同，说明消费完毕。) 

# 查询备集群的数据
curl 192.168.56.3:9200/test/_search?pretty -uadmin:admin

备集群启动后自动消费队列数据，消费完后备集群数据达到与主集群数据一致。

灾备切换

测试了这么多，终于到切换的时刻了。切换前我们判断下主系统是否短期无法修复。

如果我们判断主用系统无法短时间恢复，要执行切换。非常简单，我们直接将配置文件中定义的主备集群互换，然后重启网关程序就行了。但我们推荐在相同主机上另部署一套网关程序--网关B，先前那套用网关A指代。网关B中的配置文件把原备集群定义为主集群，原主集群定义为备集群。若要执行切换，我们先停止网关A，然后启动网关B，此时应用连接到网关(端口不变)，就把原备系统当作主系统使用，把原主系统当作备系统，也就完成了主备系统的切换。

灾备回切

当原主集群修复后，正常启动，就会从消费队列追写修复期间产生数据直到主备数据一致，同样我们可通过 INFINI Console 查看消费的进度。如果大家还是担心数据的一致性，INFINI Console 还能帮大家做[校验数据]()任务，做到数据完全一致后(文档数量及文档内容一致)，才进行回切。

回切也非常简单，停止网关B，启动网关A即可。

网关高可用

网关自带浮动 IP 模块，可进行双机热备。客户端通过 VIP 连接网关，网关出现故障时，VIP 漂移到备网关。 视频教程戳这里。

这样的优点是简单，不足是只有一个网关在线提供服务。如果想多个网关在线提供服务，则需搭配分布式消息系统一起工作，架构如下。

前端通过负载均衡将流量分散到多个在线网关，网关将消息存入分布式消息系统。此时，网关可看作无状态应用，可根据需要扩缩规模。

以上就是我介绍的ES灾备方案，是不是相当灵活了。有问题还是那句话 Call me 。

关于极限网关

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway，可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

官网文档：https://www.infinilabs.com/docs/latest/gateway

更多...

INFINI Gateway和 Console 更新发布啦！

资讯动态 • liaosy 发表了文章 • 0 个评论 • 6696 次浏览 • 2023-02-10 23:12 • 来自相关话题

INFINI Labs 产品更新发布

Hi，大家好。今天 INFINI Labs 为大家带来 2023 春节后第一波产品更新发布，欢迎大家免费下载体验使用。

INFINI Gateway v1.9.0

极限网关本次迭代带来了大量的更新如下：

Breaking changes

  • Refactoring config for ip access control
  • Disable elasticsearch metadata refresh by default
  • Update default config path from configs to config
  • Remove sample-configs, moved to dedicated integrated-testing project
  • Remove field conntime, update field @timestamp to timestamp in logging filter
  • Rename disorder to fast

Features

  • Support listen on IPv6 address
  • Add general health api
  • Add request_ip to context
  • Add badger filter plugin
  • Allow to split produce and consume messages from s3
  • Add bulk_request_throttle filter
  • Support access request context and more output options in echo filter
  • Add body_json to response context
  • Add cert config to API module, support mTLS
  • Add api to clear scroll context
  • Floating_ip support stick by priority
  • Add keystore util
  • Allow to save success bulk results in bulk_indexing processor
  • Enable watch and reload the major config file
  • Support run background job in one goroutine
  • Allow to handle async_bulk request logging
  • Add config to control cluster health check while cluster not available, set default to false
  • Allow to follow redirects in http filter, set default read and write timeout to 30s
  • Support collect instance metrics to monitoring gateway
  • Add json log format

Bug fix

  • Fix user was removed in logging filter
  • Fix incorrect message size issue, reload when files changed in disk_queue
  • Fix issue that index_diff could not finished automatically
  • Fix hostname was not well updated in filter set_request_header or set_hostname
  • Fix to check consumer’s lag instead of queue’s lag in flow_runner processor
  • Fix file not found error for disk_queue
  • Fix the delete requests was not proper handled in filter bulk_reshuffle, bulk_request_mutate and bulk_indexing processor
  • Fix memory leak caused by misuse of bytes buffer
  • Fix to handle the last request in replay processor
  • Fix url args was not updated after change
  • Fix memory leak when serving high-concurrent requests
  • Fix nil id caused error when using sliced workers in bulk_indexing processor
  • Fix index name with dot
  • Refactoring time fields for orm, skip empty time
  • Refactoring stats, allow to register extended stats
  • Fix to restart gateway entrypoint on flow change
  • Update ratio filter, fix random number, add header to ratio filter
  • Fix query parameter no_cache was not well respected in get_cache filter
  • Fix single delete request was ignored in bulk requests
  • Fix request mutate filter

Improvements

  • Remove newline in indexing_merge and json_indexing processor
  • Improve instance check, add config to disable
  • Add option skip_insecure_verify to s3 module
  • Improve instance check, enable config to disable
  • Update the way to get ctx process info, optimize memory usage
  • Improve indexing performance for bulk_indexing processor
  • Refactoring disk_queue, speedup message consumption
  • Enable segment compress for disk_queue by default
  • Skip download s3 files when s3 was not enabled
  • Add option to log warning messages for throttle filters
  • Optimize hash performance for getting primary shardID and partitionID
  • Add cache for get index routing table
  • Optimize performance for bulk response processing
  • Refactoring bulk_processor, pass meta info to payload func
  • Don’t call payload func for delete action
  • Improve queue consumer’s lag check
  • Enable prepare flat files ahead for read by default, skip unnecessary file
  • Add object pool for xxhash
  • Refactoring disk_queue, handle consumer in-flight segments in memory
  • Add config to remove duplicated newline for bulk_processor
  • Add metric timestamp in stats api
  • Improve error on routing table missing
  • Refactoring bytes buffer and object pool, expose metrics via API
  • Refactoring tasks pooling, support throttle and unified control
  • Optimize badger file size and memory usage
  • Refactoring time fields for orm, skip empty time
  • Refactoring stats, allow to register extended stats
  • Refactoring to handle bulk response results
  • Add client_session_cache_size to tls setting
  • Safety add newline to each bytes when handle bulk requests

INFINI Console v0.7.0

INFINI Console 本次迭代更新如下：

  • 新增初始化安装向导；

    image.png

  • 新增系统服务健康监控；

  • 新增 License 授权；

    image_(1).png

  • 新增索引和节点层面数据字节写入吞吐量指标(indexing bytes)；
  • 修复了 Discover 第一次加载未发起搜索请求的问题；
  • 修复了查看节点线程池指标时选择多个节点后指标不显示的问题；

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

也欢迎大家添加微信小助手(INFINI-Labs)拉群交流和学习。

640.jpeg

感谢大家的围观，祝大家周末愉快。

使用极限网关来进行 Elasticsearch 跨集群跨版本查询及所有其它请求

Elasticsearch • medcl 发表了文章 • 8 个评论 • 5178 次浏览 • 2021-10-16 11:31 • 来自相关话题

使用场景

如果你的业务需要用到有多个集群，并且版本还不一样，是不是管理起来很麻烦，如果能够通过一个 API 来进行查询就方便了，聪明的你相信已经想到了 CCS，没错用 CCS 可以实现跨集群的查询，不过 Elasticsearch 提供的 CCS 对版本有一点的限制，并且需要提前做好 mTLS，也就是需要提前配置好两个集群之间的证书互信，这个免不了要重启维护，好像有点麻烦，那么问题来咯，有没有更好的方案呢？

😁 有办法，今天我就给大家介绍一个基于极限网关的方案，极限网关的网址：http://极限网关.com/。

假设现在有两个集群，一个集群是 v2.4.6，有不少业务数据，舍不得删，里面有很多好东西 ：）还有一个集群是 v7.14.0，版本还算比较新，业务正在做的一个新的试点，没什么好东西，但是也得用 :(，现在老板们的的需求是希望通过在一个统一的接口就能访问这些数据，程序员懒得很，懂得都懂。

集群信息

  • v2.4.6 集群的访问入口地址：192.168.3.188:9202
  • v7.14.0 集群的访问入口地址：192.168.3.188:9206

这两个集群都是 http 协议的。

实现步骤

今天用到的是极限网关的 switch 过滤器：https://极限网关.com/docs/references/filters/switch/

网关下载下来就两个文件，一个主程序，一个配置文件，记得下载对应操作系统的包。

定义两个集群资源

elasticsearch:
  - name: v2
    enabled: true
    endpoint: http://192.168.3.188:9202
  - name: v7
    enabled: true
    endpoint: http://192.168.3.188:9206

上面定义了两个集群，分别命名为 v2 和 v7，待会会用到这些资源。

定义一个服务入口

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 1000
    network:
      binding: 0.0.0.0:8000
    tls:
      enabled: true

这里定义了一个名为 my_es_entry 的资源入口，并引用了一个名为 my_router 的请求转发路由，同时绑定了网卡的 0.0.0.0:8000 也就是所有本地网卡监听 IP 的 8000 端口，访问任意 IP 的 8000 端口就能访问到这个网关了。

另外老板也说了，Elasticsearch 用 HTTP 协议简直就是裸奔，通过这里开启 tls，可以让网关对外提供的是 HTTPS 协议，这样用户连接的 Elasticsearch 服务就自带 buffer 了，后端的 es 集群和网关直接可以做好网络流量隔离，集群不用动，简直完美。

为什么定义 TLS 不用指定证书，好用的软件不需要这么麻烦，就这样，不解释。

最后，通过设置 max_concurrency 为 1000，限制下并发数，避免野猴子把我们的后端的 Elasticsearch 给压挂了。

定义一个请求路由

router:
  - name: my_router
    default_flow: cross-cluster-search

这里的名称 my_router 就是表示上面的服务入口的router 参数指定的值。

另外设置一个 default_flow 来将所有的请求都转发给一个名为 cross-cluster-search 的请求处理流程，还没定义，别急，马上。

定义请求处理流程

来啦，来啦，先定义两个 flow，如下，分别名为 v2-flow 和 v7-flow，每节配置的 filter 定义了一系列过滤器，用来对请求进行处理，这里就用了一个 elasticsearch 过滤器，也就是转发请求给指定的 Elasticsearch 后端服务器，了否？

flow:
  - name: v2-flow
    filter:
      - elasticsearch:
          elasticsearch: v2
  - name: v7-flow
    filter:
      - elasticsearch:
          elasticsearch: v7

然后，在定义额外一个名为 cross-cluster-search 的 flow，如下：

  - name: cross-cluster-search
    filter:
      - switch:
          path_rules:
            - prefix: "v2:"
              flow: v2-flow
            - prefix: "v7:"
              flow: v7-flow

这个 flow 就是通过请求的路径的前缀来进行路由的过滤器，如果是 v2:开头的请求，则转发给 v2-flow 继续处理，如果是 v7: 开头的请求，则转发给 v7-flow 来处理，使用的用法和 CCS 是一样的。so easy！

对了，那是不是每个请求都需要加前缀啊，费事啊，没事，在这个 cross-cluster-search 的 filter 最后再加上一个 elasticsearch filter，前面前缀匹配不上的都会走它，假设默认都走 v7，最后完整的 flow 配置如下：

flow:
  - name: v2-flow
    filter:
      - elasticsearch:
          elasticsearch: v2
  - name: v7-flow
    filter:
      - elasticsearch:
          elasticsearch: v7
  - name: cross-cluster-search
    filter:
      - switch:
          path_rules:
            - prefix: "v2:"
              flow: v2-flow
            - prefix: "v7:"
              flow: v7-flow
      - elasticsearch:
          elasticsearch: v7

然后就没有然后了，因为就配置这些就行了。

启动网关

假设配置文件的路径为 sample-configs/cross-cluster-search.yml，运行如下命令：

➜  gateway git:(master) ✗ ./bin/gateway -config sample-configs/cross-cluster-search.yml
   ___   _   _____  __  __    __  _       
  / _ \ /_\ /__   \/__\/ / /\ \ \/_\ /\_/\
 / /_\///_\\  / /\/_\  \ \/  \/ //_\\\_ _/
/ /_\\/  _  \/ / //__   \  /\  /  _  \/ \ 
\____/\_/ \_/\/  \__/    \/  \/\_/ \_/\_/ 

[GATEWAY] A light-weight, powerful and high-performance elasticsearch gateway.
[GATEWAY] 1.0.0_SNAPSHOT, 2021-10-15 16:25:56, 3d0a1cd
[10-16 11:00:52] [INF] [app.go:228] initializing gateway.
[10-16 11:00:52] [INF] [instance.go:24] workspace: data/gateway/nodes/0
[10-16 11:00:52] [INF] [api.go:260] api listen at: http://0.0.0.0:2900
[10-16 11:00:52] [INF] [reverseproxy.go:257] elasticsearch [v7] hosts: [] => [192.168.3.188:9206]
[10-16 11:00:52] [INF] [entry.go:225] auto generating cert files
[10-16 11:00:52] [INF] [actions.go:223] elasticsearch [v2] is available
[10-16 11:00:52] [INF] [actions.go:223] elasticsearch [v7] is available
[10-16 11:00:53] [INF] [entry.go:296] entry [my_es_entry] listen at: https://0.0.0.0:8000
[10-16 11:00:53] [INF] [app.go:309] gateway is running now.

可以看到网关输出了启动成功的日志，网关服务监听在 https://0.0.0.0:8000 。

试试访问网关

直接访问网关的 8000 端口，因为是网关自签的证书，加上 -k 来跳过证书的校验，如下：

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000  
{
  "name" : "LENOVO",
  "cluster_name" : "es-v7140",
  "cluster_uuid" : "npWjpIZmS8iP_p3GK01-xg",
  "version" : {
    "number" : "7.14.0",
    "build_flavor" : "default",
    "build_type" : "zip",
    "build_hash" : "dd5a0a2acaa2045ff9624f3729fc8a6f40835aa1",
    "build_date" : "2021-07-29T20:49:32.864135063Z",
    "build_snapshot" : false,
    "lucene_version" : "8.9.0",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

正如前面配置所配置的一样，默认请求访问的就是 v7 集群。

访问 v2 集群

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000/v2:/    
{
  "name" : "Solomon O'Sullivan",
  "cluster_name" : "es-v246",
  "cluster_uuid" : "cqlpjByvQVWDAv6VvRwPAw",
  "version" : {
    "number" : "2.4.6",
    "build_hash" : "5376dca9f70f3abef96a77f4bb22720ace8240fd",
    "build_timestamp" : "2017-07-18T12:17:44Z",
    "build_snapshot" : false,
    "lucene_version" : "5.5.4"
  },
  "tagline" : "You Know, for Search"
}

查看集群信息：

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000/v2:_cluster/health\?pretty
{
  "cluster_name" : "es-v246",
  "status" : "yellow",
  "timed_out" : false,
  "number_of_nodes" : 1,
  "number_of_data_nodes" : 1,
  "active_primary_shards" : 5,
  "active_shards" : 5,
  "relocating_shards" : 0,
  "initializing_shards" : 0,
  "unassigned_shards" : 5,
  "delayed_unassigned_shards" : 0,
  "number_of_pending_tasks" : 0,
  "number_of_in_flight_fetch" : 0,
  "task_max_waiting_in_queue_millis" : 0,
  "active_shards_percent_as_number" : 50.0
}

插入一条文档：

➜  loadgen git:(master) ✗ curl-json -k   https://localhost:8000/v2:medcl/doc/1 -d '{"name":"hello world"}'
{"_index":"medcl","_type":"doc","_id":"1","_version":1,"_shards":{"total":2,"successful":1,"failed":0},"created":true}%

执行一个查询

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000/v2:medcl/_search\?q\=name:hello                
{"took":78,"timed_out":false,"_shards":{"total":5,"successful":5,"failed":0},"hits":{"total":1,"max_score":0.19178301,"hits":[{"_index":"medcl","_type":"doc","_id":"1","_score":0.19178301,"_source":{"name":"hello world"}}]}}%

可以看到，所有的请求，不管是集群的操作，还是索引的增删改查都可以，而 Elasticsearch 自带的 CCS 是只读的，只能进行查询。

访问 v7 集群

➜  loadgen git:(master) ✗ curl -k   https://localhost:8000/v7:/
{
  "name" : "LENOVO",
  "cluster_name" : "es-v7140",
  "cluster_uuid" : "npWjpIZmS8iP_p3GK01-xg",
  "version" : {
    "number" : "7.14.0",
    "build_flavor" : "default",
    "build_type" : "zip",
    "build_hash" : "dd5a0a2acaa2045ff9624f3729fc8a6f40835aa1",
    "build_date" : "2021-07-29T20:49:32.864135063Z",
    "build_snapshot" : false,
    "lucene_version" : "8.9.0",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

Kibana 里面访问

完全没问题，有图有真相：

Jietu20211016-114041.jpg

其他操作也类似，就不重复了。

完整的配置

path.data: data
path.logs: log

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 10000
    network:
      binding: 0.0.0.0:8000
    tls:
      enabled: true

flow:
  - name: v2-flow
    filter:
      - elasticsearch:
          elasticsearch: v2
  - name: v7-flow
    filter:
      - elasticsearch:
          elasticsearch: v7
  - name: cross-cluster-search
    filter:
      - switch:
          path_rules:
            - prefix: "v2:"
              flow: v2-flow
            - prefix: "v7:"
              flow: v7-flow
      - elasticsearch:
          elasticsearch: v7

router:
  - name: my_router
    default_flow: cross-cluster-search

elasticsearch:
  - name: v2
    enabled: true
    endpoint: http://192.168.3.188:9202
  - name: v7
    enabled: true
    endpoint: http://192.168.3.188:9206

小结

好了，今天给大家分享的如何使用极限网关来进行 Elasticsearch 跨集群跨版本的操作就到这里了，希望大家周末玩的开心。😁

更多...

集群中gateway.*参数有不同值会导致什么结果？

回复

Elasticsearch • medcl 回复了问题 • 2 人关注 • 1 个回复 • 4001 次浏览 • 2017-07-24 10:59 • 来自相关话题

更多...

ES 调优帖：Gateway 批量写入性能优化实践

Elasticsearch • INFINI Labs 小助手 发表了文章 • 0 个评论 • 5006 次浏览 • 2025-08-06 17:32 • 来自相关话题

背景：bulk 优化的应用

在 ES 的写入优化里，bulk 操作被广泛地用于批量处理数据。bulk 操作允许用户一次提交多个数据操作，如索引、更新、删除等，从而提高数据处理效率。bulk 操作的实现原理是，将数据操作请求打包成 HTTP 请求，并批量提交给 Elasticsearch 服务器。这样，Elasticsearch 服务器就可以一次处理多个数据操作，从而提高处理效率。

这种优化的核心价值在于减少了网络往返的次数和连接建立的开销。每一次单独的写入操作都需要经历完整的请求-响应周期，而批量写入则是将多个操作打包在一起，用一次通信完成原本需要多次交互的工作。这不仅仅节省了时间，更重要的是释放了系统资源，让服务器能够专注于真正的数据处理，而不是频繁的协议握手和状态维护。

这样的批量请求的确是可以优化写入请求的效率，让 ES 集群获得更多的资源去做写入请求的集中处理。但是除了客户端与 ES 集群的通讯效率优化，还有其他中间过程能优化么？

Gateway 的优化点

bulk 的优化理念是将日常零散的写入需求做集中化的处理，尽量减低日常请求的损耗，完成资源最大化的利用。简而言之就是“好钢用在刀刃上”。

但是 ES 在收到 bulk 写入请求后，也是需要协调节点根据文档的 id 计算所属的分片来将数据分发到对应的数据节点的。这个过程也是有一定损耗的，如果 bulk 请求中数据分布的很散，每个分片都需要进行写入，原本 bulk 集中写入的需求优势则还是没有得到最理想化的提升。

gateway 的写入加速则对 bulk 的优化理念的最大化补全。

gateway 可以本地计算每个索引文档对应后端 Elasticsearch 集群的目标存放位置，从而能够精准的进行写入请求定位。

在一批 bulk 请求中，可能存在多个后端节点的数据，bulk_reshuffle 过滤器用来将正常的 bulk 请求打散，按照目标节点或者分片进行拆分重新组装，避免 Elasticsearch 节点收到请求之后再次进行请求分发， 从而降低 Elasticsearch 集群间的流量和负载，也能避免单个节点成为热点瓶颈，确保各个数据节点的处理均衡，从而提升集群总体的索引吞吐能力。

整理的优化思路如下图：

优化实践

那我们来实践一下，看看 gateway 能提升多少的写入。

这里我们分 2 个测试场景：

  1. 基础集中写入测试，不带文档 id，直接批量写入。这个场景更像是日志或者监控数据采集的场景。
  2. 带文档 id 的写入测试，更偏向搜索场景或者大数据批同步的场景。

2 个场景都进行直接写入 ES 和 gateway 转发 ES 的效率比对。

测试材料除了需要备一个网关和一套 es 外，其余的内容如下：

测试索引 mapping 一致，名称区分：

PUT gateway_bulk_test
{
  "settings": {
    "number_of_shards": 6,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "timestamp": {
        "type": "date",
        "format": "strict_date_optional_time"
      },
      "field1": {
        "type": "keyword"
      },
      "field2": {
        "type": "keyword"
      },
      "field3": {
        "type": "keyword"
      },
      "field4": {
        "type": "integer"
      },
      "field5": {
        "type": "keyword"
      },
      "field6": {
        "type": "float"
      }
    }
  }
}

PUT bulk_test
{
  "settings": {
    "number_of_shards": 6,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "timestamp": {
        "type": "date",
        "format": "strict_date_optional_time"
      },
      "field1": {
        "type": "keyword"
      },
      "field2": {
        "type": "keyword"
      },
      "field3": {
        "type": "keyword"
      },
      "field4": {
        "type": "integer"
      },
      "field5": {
        "type": "keyword"
      },
      "field6": {
        "type": "float"
      }
    }
  }
}

gateway 的配置文件如下：

path.data: data
path.logs: log

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 200000
    network:
      binding: 0.0.0.0:8000

flow:
  - name: async_bulk
    filter:
      - bulk_reshuffle:
          when:
            contains:
              _ctx.request.path: /_bulk
          elasticsearch: prod
          level: node
          partition_size: 1
          fix_null_id: true
      - elasticsearch:
          elasticsearch: prod #elasticsearch configure reference name
          max_connection_per_node: 1000 #max tcp connection to upstream, default for all nodes
          max_response_size: -1 #default for all nodes
          balancer: weight
          refresh: # refresh upstream nodes list, need to enable this feature to use elasticsearch nodes auto discovery
            enabled: true
            interval: 60s
          filter:
            roles:
              exclude:
                - master

router:
  - name: my_router
    default_flow: async_bulk

elasticsearch:
  - name: prod
    enabled: true
    endpoints:
      - https://127.0.0.1:9221
      - https://127.0.0.1:9222
      - https://127.0.0.1:9223
    basic_auth:
      username: admin
      password: admin

pipeline:
  - name: bulk_request_ingest
    auto_start: true
    keep_running: true
    retry_delay_in_ms: 1000
    processor:
      - bulk_indexing:
          max_connection_per_node: 100
          num_of_slices: 3
          max_worker_size: 30
          idle_timeout_in_seconds: 10
          bulk:
            compress: false
            batch_size_in_mb: 10
            batch_size_in_docs: 10000
          consumer:
            fetch_max_messages: 100
          queue_selector:
            labels:
              type: bulk_reshuffle

测试脚本如下：

#!/usr/bin/env python3
"""
ES Bulk写入性能测试脚本

"""

import hashlib
import json
import random
import string
import time
from typing import List, Dict, Any

import requests
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
import urllib3

# 禁用SSL警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class ESBulkTester:
    def __init__(self):
        # 配置变量 - 可修改
        self.es_configs = [
            {
                "name": "ES直连",
                "url": "https://127.0.0.1:9221",
                "index": "bulk_test",
                "username": "admin",  # 修改为实际用户名
                "password": "admin",  # 修改为实际密码
                "verify_ssl": False  # HTTPS需要SSL验证
            },
            {
                "name": "Gateway代理",
                "url": "http://localhost:8000",
                "index": "gateway_bulk_test",
                "username": None,  # 无需认证
                "password": None,
                "verify_ssl": False
            }
        ]
        self.batch_size = 10000  # 每次bulk写入条数
        self.log_interval = 100000  # 每多少次bulk写入输出日志

        # ID生成规则配置 - 前2位后5位
        self.id_prefix_start = 1
        self.id_prefix_end = 999      # 前3位: 01-999
        self.id_suffix_start = 1
        self.id_suffix_end = 9999   # 后4位: 0001-9999

        # 当前ID计数器
        self.current_prefix = self.id_prefix_start
        self.current_suffix = self.id_suffix_start

    def generate_id(self) -> str:
        """生成固定规则的ID - 前2位后5位"""
        id_str = f"{self.current_prefix:02d}{self.current_suffix:05d}"

        # 更新计数器
        self.current_suffix += 1
        if self.current_suffix > self.id_suffix_end:
            self.current_suffix = self.id_suffix_start
            self.current_prefix += 1
            if self.current_prefix > self.id_prefix_end:
                self.current_prefix = self.id_prefix_start

        return id_str

    def generate_random_hash(self, length: int = 32) -> str:
        """生成随机hash值"""
        random_string = ''.join(random.choices(string.ascii_letters + string.digits, k=length))
        return hashlib.md5(random_string.encode()).hexdigest()

    def generate_document(self) -> Dict[str, Any]:
        """生成随机文档内容"""
        return {
            "timestamp": datetime.now().isoformat(),
            "field1": self.generate_random_hash(),
            "field2": self.generate_random_hash(),
            "field3": self.generate_random_hash(),
            "field4": random.randint(1, 1000),
            "field5": random.choice(["A", "B", "C", "D"]),
            "field6": random.uniform(0.1, 100.0)
        }

    def create_bulk_payload(self, index_name: str) -> str:
        """创建bulk写入payload"""
        bulk_data = []

        for _ in range(self.batch_size):
            #doc_id = self.generate_id()
            doc = self.generate_document()

            # 添加index操作
            bulk_data.append(json.dumps({
                "index": {
                    "_index": index_name,
            #        "_id": doc_id
                }
            }))
            bulk_data.append(json.dumps(doc))

        return "\n".join(bulk_data) + "\n"

    def bulk_index(self, config: Dict[str, Any], payload: str) -> bool:
        """执行bulk写入"""
        url = f"{config['url']}/_bulk"
        headers = {
            "Content-Type": "application/x-ndjson"
        }

        # 设置认证信息
        auth = None
        if config.get('username') and config.get('password'):
            auth = (config['username'], config['password'])

        try:
            response = requests.post(
                url,
                data=payload,
                headers=headers,
                auth=auth,
                verify=config.get('verify_ssl', True),
                timeout=30
            )
            return response.status_code == 200
        except Exception as e:
            print(f"Bulk写入失败: {e}")
            return False

    def refresh_index(self, config: Dict[str, Any]) -> bool:
        """刷新索引"""
        url = f"{config['url']}/{config['index']}/_refresh"

        # 设置认证信息
        auth = None
        if config.get('username') and config.get('password'):
            auth = (config['username'], config['password'])

        try:
            response = requests.post(
                url,
                auth=auth,
                verify=config.get('verify_ssl', True),
                timeout=10
            )
            success = response.status_code == 200
            print(f"索引刷新{'成功' if success else '失败'}: {config['index']}")
            return success
        except Exception as e:
            print(f"索引刷新失败: {e}")
            return False

    def run_test(self, config: Dict[str, Any], round_num: int, total_iterations: int = 100000):
        """运行性能测试"""
        test_name = f"{config['name']}-第{round_num}轮"
        print(f"\n开始测试: {test_name}")
        print(f"ES地址: {config['url']}")
        print(f"索引名称: {config['index']}")
        print(f"认证: {'是' if config.get('username') else '否'}")
        print(f"每次bulk写入: {self.batch_size}条")
        print(f"总计划写入: {total_iterations * self.batch_size}条")
        print("-" * 50)

        start_time = time.time()
        success_count = 0
        error_count = 0

        for i in range(1, total_iterations + 1):
            payload = self.create_bulk_payload(config['index'])

            if self.bulk_index(config, payload):
                success_count += 1
            else:
                error_count += 1

            # 每N次输出日志
            if i % self.log_interval == 0:
                elapsed_time = time.time() - start_time
                rate = i / elapsed_time if elapsed_time > 0 else 0
                print(f"已完成 {i:,} 次bulk写入, 耗时: {elapsed_time:.2f}秒, 速率: {rate:.2f} bulk/秒")

        end_time = time.time()
        total_time = end_time - start_time
        total_docs = total_iterations * self.batch_size

        print(f"\n{test_name} 测试完成!")
        print(f"总耗时: {total_time:.2f}秒")
        print(f"成功bulk写入: {success_count:,}次")
        print(f"失败bulk写入: {error_count:,}次")
        print(f"总文档数: {total_docs:,}条")
        print(f"平均速率: {success_count/total_time:.2f} bulk/秒")
        print(f"文档写入速率: {total_docs/total_time:.2f} docs/秒")
        print("=" * 60)

        return {
            "test_name": test_name,
            "config_name": config['name'],
            "round": round_num,
            "es_url": config['url'],
            "index": config['index'],
            "total_time": total_time,
            "success_count": success_count,
            "error_count": error_count,
            "total_docs": total_docs,
            "bulk_rate": success_count/total_time,
            "doc_rate": total_docs/total_time
        }

    def run_comparison_test(self, total_iterations: int = 10000):
        """运行双地址对比测试"""
        print("ES Bulk写入性能测试开始")
        print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("=" * 60)

        results = []
        rounds = 2  # 每个地址测试2轮

        # 循环测试所有配置
        for config in self.es_configs:
            print(f"\n开始测试配置: {config['name']}")
            print("*" * 40)

            for round_num in range(1, rounds + 1):
                # 运行测试
                result = self.run_test(config, round_num, total_iterations)
                results.append(result)

                # 每轮结束后刷新索引
                print(f"\n第{round_num}轮测试完成，执行索引刷新...")
                self.refresh_index(config)

                # 重置ID计数器
                if round_num == 1:
                    # 第1轮：使用初始ID范围（新增数据）
                    print("第1轮：新增数据模式")
                else:
                    # 第2轮：重复使用相同ID（更新数据模式）
                    print("第2轮：数据更新模式，复用第1轮ID")
                    self.current_prefix = self.id_prefix_start
                    self.current_suffix = self.id_suffix_start

                print(f"{config['name']} 第{round_num}轮测试结束\n")

        # 输出对比结果
        print("\n性能对比结果:")
        print("=" * 80)

        # 按配置分组显示结果
        config_results = {}
        for result in results:
            config_name = result['config_name']
            if config_name not in config_results:
                config_results[config_name] = []
            config_results[config_name].append(result)

        for config_name, rounds_data in config_results.items():
            print(f"\n{config_name}:")
            total_time = 0
            total_bulk_rate = 0
            total_doc_rate = 0

            for round_data in rounds_data:
                print(f"  第{round_data['round']}轮:")
                print(f"    耗时: {round_data['total_time']:.2f}秒")
                print(f"    Bulk速率: {round_data['bulk_rate']:.2f} bulk/秒")
                print(f"    文档速率: {round_data['doc_rate']:.2f} docs/秒")
                print(f"    成功率: {round_data['success_count']/(round_data['success_count']+round_data['error_count'])*100:.2f}%")

                total_time += round_data['total_time']
                total_bulk_rate += round_data['bulk_rate']
                total_doc_rate += round_data['doc_rate']

            avg_bulk_rate = total_bulk_rate / len(rounds_data)
            avg_doc_rate = total_doc_rate / len(rounds_data)

            print(f"  平均性能:")
            print(f"    总耗时: {total_time:.2f}秒")
            print(f"    平均Bulk速率: {avg_bulk_rate:.2f} bulk/秒")
            print(f"    平均文档速率: {avg_doc_rate:.2f} docs/秒")

        # 整体对比
        if len(config_results) >= 2:
            config_names = list(config_results.keys())
            config1_avg = sum([r['bulk_rate'] for r in config_results[config_names[0]]]) / len(config_results[config_names[0]])
            config2_avg = sum([r['bulk_rate'] for r in config_results[config_names[1]]]) / len(config_results[config_names[1]])

            if config1_avg > config2_avg:
                faster = config_names[0]
                rate_diff = config1_avg - config2_avg
            else:
                faster = config_names[1]
                rate_diff = config2_avg - config1_avg

            print(f"\n整体性能对比:")
            print(f"{faster} 平均性能更好，bulk速率高 {rate_diff:.2f} bulk/秒")
            print(f"性能提升: {(rate_diff/min(config1_avg, config2_avg)*100):.1f}%")

def main():
    """主函数"""
    tester = ESBulkTester()

    # 运行测试（每次bulk 1万条，300次bulk = 300万条文档）
    tester.run_comparison_test(total_iterations=300)

if __name__ == "__main__":
    main()

1. 日志场景：不带 id 写入

测试条件：

  1. bulk 写入数据不带文档 id
  2. 每批次 bulk 10000 条数据，总共写入 30w 数据

这里把

反馈结果：

性能对比结果:
================================================================================

ES直连:
  第1轮:
    耗时: 152.29秒
    Bulk速率: 1.97 bulk/秒
    文档速率: 19699.59 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 152.29秒
    平均Bulk速率: 1.97 bulk/秒
    平均文档速率: 19699.59 docs/秒

Gateway代理:
  第1轮:
    耗时: 115.63秒
    Bulk速率: 2.59 bulk/秒
    文档速率: 25944.35 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 115.63秒
    平均Bulk速率: 2.59 bulk/秒
    平均文档速率: 25944.35 docs/秒

整体性能对比:
Gateway代理 平均性能更好，bulk速率高 0.62 bulk/秒
性能提升: 31.7%

2. 业务场景：带文档 id 的写入

测试条件：

  1. bulk 写入数据带有文档 id，两次测试写入的文档 id 生成规则一致且重复。
  2. 每批次 bulk 10000 条数据，总共写入 30w 数据

这里把 py 脚本中 第 99 行 和 第 107 行的注释打开。

反馈结果：

性能对比结果:
================================================================================

ES直连:
  第1轮:
    耗时: 155.30秒
    Bulk速率: 1.93 bulk/秒
    文档速率: 19317.39 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 155.30秒
    平均Bulk速率: 1.93 bulk/秒
    平均文档速率: 19317.39 docs/秒

Gateway代理:
  第1轮:
    耗时: 116.73秒
    Bulk速率: 2.57 bulk/秒
    文档速率: 25700.06 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 116.73秒
    平均Bulk速率: 2.57 bulk/秒
    平均文档速率: 25700.06 docs/秒

整体性能对比:
Gateway代理 平均性能更好，bulk速率高 0.64 bulk/秒
性能提升: 33.0%

小结

不管是日志场景还是业务价值更重要的大数据或者搜索数据同步场景， gateway 的写入加速都能平稳的节省 25%-30% 的写入耗时。

关于极限网关（INFINI Gateway）

INFINI Gateway 是一个开源的面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway，可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

官网文档：https://docs.infinilabs.com/gateway
开源地址：https://github.com/infinilabs/gateway

作者：金多安，极限科技（INFINI Labs）搜索运维专家，Elastic 认证专家，搜索客社区日报责任编辑。一直从事与搜索运维相关的工作，日常会去挖掘 ES / Lucene 方向的搜索技术原理，保持搜索相关技术发展的关注。
原文：https://infinilabs.cn/blog/2025/gateway-bulk-write-performance-optimization/

INFINI Labs 产品更新 | Coco AI – 增强 AI 搜索、API 管理与性能优化等

资讯动态 • INFINI Labs 小助手 发表了文章 • 0 个评论 • 3419 次浏览 • 2025-03-15 18:21 • 来自相关话题

release

INFINI Labs 产品更新发布！此次更新涵盖 Coco AI 、Easysearch 等产品多项重要升级，重点提升 AI 搜索能力、易用性及企业级优化。

  • Coco AI v0.2 作为 开源、跨平台的 AI 搜索工具，新增 APP 自动更新提示、API Token 管理、文档处理优化 等功能。
  • INFINI Easysearch v1.11.1 集成 AI 向量搜索，优化查询聚合能力。
  • INFINI Console、Gateway、Agent、Loadgen、Framework 关键问题修复，优化 WebSocket 处理与整体用户体验。
  • Coco AI Server 增强 WebSocket 会话处理、支持 RAG 会话和动态配置，并新增图形化管理界面。

Coco AI v0.2

Coco AI 是一个完全开源、跨平台的统一 AI 搜索与效率工具，能够连接并搜索多种数据源，包括应用程序、文件、谷歌网盘、Notion、语雀、Hugo 等本地与云端数据。通过接入 DeepSeek 等大模型，Coco AI 实现了智能化的个人知识库管理，注重隐私，支持私有部署，帮助用户快速、智能地访问信息。

Coco AI 本次详细更新记录如下：

Coco AI 客户端 v0.2.1

功能更新

  • 支持 APP 应用内更新提示并可自动更新

问题修复

  • 修复融合搜索包含已禁用服务器的问题
  • 修复版本类型不正确：应为字符串而不是 u32
  • 修复聊天推送结束的判断类型不准确问题

优化改进

  • 重构了聊天组件
  • 添加服务链接展示
  • 优化了聊天滚动效果和聊天数据渲染效果
  • 设置聊天窗口最小宽度 & 移除输入框背景
  • 移除废弃的选中功能 & 添加选择隐藏 APP 功能
  • Websocket 超时增加到 2 分钟

Coco AI 服务端 v0.2.2

功能更新

  • 新增图形化管理界面
  • 新增数据源下文档创建 API
  • 新增文件上传相关 API
  • 新增 API TOKEN 管理相关 API
  • 数据源同步支持动态配置时间间隔
  • 支持动态更新服务端设置
  • 支持动态更新大模型相关设置
  • 新增 RAG 聊天会话处理
  • 新增联网搜索能力
  • 支持对接 Deepseek 大模型
  • 新增文档预处理 Processor

问题修复

  • 修复 Google Drive Connector 缺少文件报错

优化改进

  • 优化聊天会话功能
  • 优化 Websocket 会话管理
  • 优化登录退出接口
  • 保存 Notion 其它内容到 Payload 字段
  • 完善后台任务退出机制
  • 优化默认索引模版和查询模版

INFINI Easysearch v1.11.1

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

Easysearch 本次更新如下：

功能更新

  • 新增 AI 模块，集成 Ollama embedding API，支持文本向量化

问题修复

  • 修复 DateRange 聚合在 Rollup 查询中无法正确合并的问题

优化改进

针对用户使用体验进行了多项改进，包括：

  • 弃用 KNN 模块中的 index.knn 配置项，（此配置项和其他功能经常发生冲突） 简化配置逻辑，该配置项将在后续版本中移除
  • 将 KNN 搜索功能从插件形式集成为内置功能，无需额外安装即可使用
  • 将跨集群复制（CCR）功能从插件形式集成为内置功能，开箱即用
  • 优化索引配置更新验证：增加非动态配置项的值比对，避免误报

INFINI Console v1.29.1

INFINI Console 是一款开源的非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验：

http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

问题修复

  • 修复 agentless 模式下计算索引级别实时 QPS 不准确的问题

INFINI Gateway v1.29.1

INFINI Gateway 是一个开源的面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

优化改进

INFINI Agent v1.29.1

INFINI Agent 负责采集和上传 Elasticsearch, Easysearch, Opensearch 集群的日志和指标信息，通过 INFINI Console 管理，支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。

Agent 本次更新如下：

优化改进

INFINI Loadgen v1.29.1

INFINI Loadgen 是一款开源的专为 Easysearch、Elasticsearch、OpenSearch 设计的轻量级性能测试工具。

Loadgen 本次更新如下：

优化改进

INFINI Framework v1.1.4

INFINI Framework 是 INFINI Labs 基于 Golang 的产品的核心基础，已开源。该框架以开发者为中心设计，简化了构建高性能、可扩展且可靠的应用程序的过程。

Framework 本次更新如下：

功能更新

  • 添加配置选项，以在 WebSocket 连接期间禁用回显消息
  • 允许在 WebSocket 连接/断开时注册回调函数
  • 为 API 添加可选的登录验证配置

优化改进

  • 停止任务后取消任务
  • 回调发生错误时关闭 WebSocket 连接

更多详情请查看以下详细的 Release Notes 或联系我们的技术支持团队！

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

下载地址： https://infinilabs.cn/download

邮件：hello@infini.ltd

电话：(+86) 400-139-9200

Discord：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://infinilabs.cn

INFINI Labs 产品更新 | Console/Gateway/Agent 等产品开源发布首个版本

资讯动态 • INFINI Labs 小助手 发表了文章 • 1 个评论 • 3633 次浏览 • 2024-12-27 16:11 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Easysearch v1.9.0、Gateway、Console、Agent、Loadgen v1.27.0。本次各产品更新了很多亮点功能，如 Easysearch 新增 rollup 功能，优化了多版本兼容配置；Console/Gateway/Agent/Loadgen 及 Framework 开源后，发布首个重大更新版本，支持过期元数据删除，指标图表懒加载，指标采集协程优化等等，欢迎大家下载体验。

INFINI Easysearch v1.9.0

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。详情见：https://infinilabs.cn

Easysearch 本次更新如下：

Improvements

  • 发布 rollup 功能

    • 支持自动对 rollup 索引进行滚动，无需外部触发
    • 支持 avg sum max min value_count percentiles 指标类型的聚合
    • 支持 terms 聚合
    • 支持对指标聚合进行 Pipeline 聚合
    • 支持聚合前先对数据进行过滤
    • 进行聚合查询时支持直接搜索原始索引，不用更改搜索代码
  • 增加适配 logstash 8.x 的请求 header
  • _cat/templates 增加 lifecycle 和 rollover 列的展示

Bug fix

  • 修复 rest-api template 测试错误

INFINI Console v1.27.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。开源地址：https://github.com/infinilabs/console

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Improvements

  • 代码开源，统一采用 Github 仓库进行开发
  • 指标采集优化，由原来的单一协程采集调整为每个注册的集群有单独的协程进行采集
  • 指标监控页面图表展示采用懒加载、单个图表独立加载，增强用户体验
  • 通用时间控件增加超时时间设置
  • 集群选择控件增加注册、刷新功能
  • 提供指标采集状态
  • 表格控件排版优化

Bug fix

  • 修复集群元数据更新不及时问题
  • 修复帮助文档等链接不正确问题
  • 修复节点、索引数据因随机 id 出现重复记录问题
  • 修复 Runtime、Agent 实例编辑页面出错问题
  • 修复集群、节点、索引、分片元数据无 Loading 问题
  • 修复索引健康状态指标采集失败问题
  • 修复个别菜单列未国际化问题

INFINI Gateway v1.27.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。开源地址：https://github.com/infinilabs/gateway

Gateway 本次更新如下：

Improvements

  • 调整队列消费者 slice 默认配置为 1

Bug fix

  • 修复缓存数据丢失导致队列无法消费问题
  • 同步更新 Framework 修复的一些已知问题

INFINI Agent v1.27.0

INFINI Agent 是 INFINI Console 的一个可选探针组件，负责采集和上传集群指标和日志等信息，并可通过 Console 管理。Agent 支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。开源地址：https://github.com/infinilabs/agent

探针 Agent 本次更新如下：

Improvements

  • 与 INFINI Console 统一版本号
  • 同步更新 Framework 修复的已知问题
  • 支持 K8S 环境指标采集

INFINI Loadgen v1.27.0

INFINI Loadgen 是一款轻量、无依赖的 Eaysearch/Elasticsearch/OpenSearch 性能压测工具，支持参数模板化配置，支持压测端均衡流量控制，可以模拟高并发请求。开源地址：https://github.com/infinilabs/loadgen

Loadgen 本次更新如下：

Improvements

  • 保持与 Console 相同版本
  • 同步更新 Framework 修复的已知问题

Bug fix

  • 修复 API 接口测试逻辑异常问题

INFINI Framework

INFINI Framework 是 INFINI Labs 各产品依赖的核心公共代码库。开源地址：https://github.com/infinilabs/framework

Improvements

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

下载地址： https://infinilabs.cn/download

邮件：hello@infini.ltd

电话：(+86) 400-139-9200

Discord：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://infinilabs.cn

INFINI Labs 助力开源与教育：免费许可证计划全面升级

开源项目 • INFINI Labs 小助手 发表了文章 • 0 个评论 • 3348 次浏览 • 2024-06-19 17:21 • 来自相关话题

免费许可证计划

在数字化浪潮席卷全球的今天，INFINI Labs 深刻认识到开源项目和教育机构在技术创新与人才培养中的核心作用。因此，我们郑重推出全新升级的免费许可证计划，旨在全球范围内为开源社区和教育界提供有力支持，共同推动软件生态的繁荣与进步。

一、产品实力与荣誉

1.INFINI Pizza：实时搜索的新纪元

INFINI Pizza

  • 在第十三届“数据技术嘉年华”（DTC2024）上，INFINI Labs 发布了划时代的搜索引擎——INFINI Pizza，标志着搜索型数据库迈入实时搜索的新纪元。
  • INFINI Pizza 凭借先进的设计理念与架构，以及独有的专利技术，实现了对海量数据的无限伸缩，提供高效、准确的实时数据搜索能力

2.行业标杆案例

  • INFINI Labs 荣获中国信通院大数据“星河”标杆案例，其中移动云搜索数据库案例更是荣选为数据库标杆案例。
  • 该案例基于移动云 Easysearch 数据库，通过创新的多集群协同模式，实现了数据高性能存取，展现出极高的经济价值与社会价值。

3.国家发明专利认可

  • INFINI Labs 的多项自主研发技术获得国家发明专利授权，这些成果彰显了公司在大数据领域的技术实力与创新精神。

二、品牌与行业地位

  • INFINI Labs 作为搜索型数据库产品领域的领军企业，积极参与行业标准的制定与推动。

  • 其核心产品 INFINI Easysearch 荣获信通院首批可信搜索型数据库产品证书，再次印证了公司在行业中的领先地位。

三、产品介绍

Products

  1. INFINI Easysearch：作为 Elasticsearch 的国产化替代方案，提供高度兼容性与卓越性能，满足企业级需求。

  2. INFINI Console：轻量级多集群、跨版本搜索基础设施统一管控平台，助力企业高效管理搜索集群。

  3. INFINI Gateway：专为 Elasticsearch 打造的高性能应用网关，提供丰富的功能特性与卓越性能。

  4. INFINI Loadgen：支持多种搜索引擎的轻量级压测工具，为企业提供强大的数据加载与测试能力。

  5. INFINI Pizza：引领实时搜索时代的新星，为企业提供高效、准确的实时数据搜索解决方案。

四、免费许可证计划

1.教育机构学术许可证

  • 面向全球公立或私立学校、职业学校、大学等教育机构，提供非商业用途的软件使用许可。

  • 有效期一年，符合条件的教育机构可继续申请。

2.开源项目许可证

  • 面向非商业开源项目开发者，要求项目拥有活跃社区并在其官网添加 INFINI Labs 的链接。

  • 许可证免费，有效期一年，符合条件的项目可继续申请。

五、申请方式

符合条件的开源项目和教育机构可通过访问 INFINI Labs 官方网站，轻松提交申请，我们将尽快审核并回复。

申请链接：https://infinilabs.cn/community

六、结语

INFINI Labs 以全新升级的免费许可证计划为契机，与全球开源社区和教育界携手合作，共同推动软件生态的创新与发展。让我们共同迎接更加美好的未来！

七、关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.cn

联系我们

原文：https://infinilabs.cn/blog/2024/infini-labs-for-open-source-education-the-free-license-program-upgrades/

极限网关助力好未来 Elasticsearch 容器化升级

Elasticsearch • INFINI Labs 小助手 发表了文章 • 0 个评论 • 4367 次浏览 • 2024-06-12 15:01 • 来自相关话题

极限网关在好未来的最佳实践案例，轻松扛住日增百 TB 数据的流量，助力 ES 从物理机到云原生架构的改造，实现了流控、请求分析、安全管理、无缝迁移等场景。一次完美的客户体验~

背景

物理机架构时代

2022 年，好未来整个日志 Elasticsearch 拥有数十套服务集群，几百台物理机。这么多台机器耗费成本非常高，而且还要花费很大精力去维护。在人力资源有限情况下，存在非常多的弊端，运行成本高，不仅是机器折旧还有机柜等费用。

流量特征

这是来自某个业务线，如下图 1，真实流量，潮汐性非常明显。好未来有很多条业务线，几乎跟这个趋势都一致的，除了个别业务有“续报”、“开课”等活动特殊情况。潮汐性带来的问题就是高峰期 CPU、内存资源是可以消耗很高；低峰期资源使用量非常低，由于是物理架构，这些资源无法给其他业务线共享。

图1

图1

降本增效-容器化改造原动力

日志服务对成本的空前的压力促使我们推进 Elasticsearch 进行架构改造；如何改造，改造成什么样子，这两个问题一直是推进改造原动力。业界能够同时对水平扩展和垂直扩展就是 K8S，我们开始对 Elasticsearch 改造成能在 K8S 上运行进行探索，从而提升 CPU、内存利用率。

物理机时代，没办法把资源动态的扩缩，动态调配，资源隔离，单靠人力操作调度成本太高，几乎无法完成；集群对内存资源需求要比 CPU 资源大很多，由于机器型号配置是固定的，无法“定制”，这也会导致成本居高不下。所以，无论从那个方面来讲，容器化优势非常明显，既能够优化成本，也能够降低运维复杂度。

ES 容器化改造

进行架构升级重点难点- API 服务

改造过程，我们遇到了很多问题，比如容器 ES 版本和物理机 ES 版本不一致，如何让 ES API 能够兼容不同的 ES 版本，由于版本的不兼容，导致无法直接使用原有的 tribenode 进行服务，怎么提供一个高可用的 Elasticsearch API 服务。我们考虑到多个方面，比如使用官方推荐的 proxy 模式、第三方服务等进行选择，经过多方面对比，选择了极限网关 进行 tribenode 替换。

原始 ES API 服务痛点

  • API 访问没有流量控制
  • 可观测性差，而且稳定性一般
  • 版本兼容性差

物理机时代 API 架构

在物理机时代 ES 集群，API 架构如图 2，可以明显看到 tribe node 对所有 ES 集群的“侵入性”是非常大的，这就带来了很多问题，比较严重的就是单个集群对 ES tribenode 的影响和版本升级带来的不兼容问题。

图2

图2

混合时代 API 架构

通过图 3，我们可以看到，极限网关对于版本兼容性很好，能够适配不同的版本。因此，最终选择极限网关作为下一代 ES API 服务方。

图3

图3

里程碑：全部 ES 集群容器化

在 2023 年 3 月份，通过 Elastic 官方 ECK 模式，完成全部日志 ES 集群容器化改造，拥有数百节点，1PB+ 数据存储，每日新增数据 100T 左右。紧接着，除了日志服务外，同时支持了好未来多条业务线。

图4

图4

极限网关实践

下面主要讲述了，为什么选择极限网关，以及极限网关在好未来落地、应用这些内容。

为什么选择极限网关？

学习成本低

我们可以从文档中看到极限网关，其架构简洁，语法简单，直观易懂。学习成本比较低，上手非常快，对新手友好。

性能强悍

经过压测，发现极限网关速度非常快，且针对 Elasticsearch 做了非常细致的优化，能成倍提升写入和查询的速度。

安全性高

支持多种认证方式，最简单的账号密码认证，可以给自定义多个账户密码，大大简化了 Elasticsearch 的安全设置，同时，还可以支持 LDAP 安全验证。

跨版本支持

我们容器化改造过程需要兼容不同版本的 Elasticsrearch，极限网关针对不同的 Elasticsearch 版本做了兼容和针对性处理，能够让业务代码无缝的进行适配，后端 Elasticsearch 集群版本升级能够做到无缝过渡，降低版本升级和数据迁移的复杂度，非常匹配我们的业务场景。

灵活可扩展

可灵活对每个请求进行干预和路由，支持路由的智能学习，内置丰富的过滤器，通过配置动态修改每个请求的处理逻辑，也支持通过插件来进行扩展，满足我们对流量的控制，尤其是限流、用户、IP 等这些功能非常实用。

启用安全策略-为 API 服务保驾护航

痛点

在升级之前使用 tribe 作为 API 服务提供后端，几乎相当于裸奔，没有任何认证策略；另外，tribe 本身的稳定性也有问题，官方在新版本逐渐废弃这种 CCS（跨集群搜索），期间出现多次服务崩溃。

极限网关解决问题

极限网关通过，“basic_auth” 插件，提供最基本的安全校验，使用起来非常方便；同时，极限网关提供 LDAP 插件，可以接入公共的 LDAP 服务，对所有的访问用户进行校验，安全策略对所有的用户生效，不用担心因为 IP 问题泄漏数据等。

强大的过滤功能

在使用 ES 集群过程中，许多场景，需要对请求进行控制、限制等操作。在这方便，感受到了极限网关强大的产品力。比如下面的两个场景

对异常流量进行限流

  • 支持对 IP 限流
  • 支持对 hostname 限流
  • 支持 header 限流

对异常用户进行封禁

当 Elasticsearch 是通过 Basic Auth 或者 LDAP 方式来进行身份认证的时候，request_user_filter 过滤器可用来按请求的用户名信息来进行过滤。操作起来也非常简单，只需要 request_user_filter 这一个过滤器。

- request_user_filter:
    include:
        - "elastic"
    exclude：
        - "Ryan"

总结来讲，主要有这些方面的功能：

图5

图5

优秀的可观测性

痛点

改造前经常为看不到直观的数据指标感到头疼，查看指标需要多个地方同时打开，去筛选，查找，非常繁琐，付出的成本非常大。为此，大家都再考虑如何优化这种情况，无奈优先级比较低，一直没有真正的投入时间去优化这块。

完美解决

使用了极限网关，通过收集请求日志，非常清晰的收集到想要的数据，具体如下：

  • 总体方面：
    • 流量曲线
    • 状态码占比
    • 缓存统计
    • 每台网关请求流量
  • 细节方面：
    • 打印每次请求语句
    • 可以查看请求到具体 ES 节点流量
    • 可以查看过滤器的列表

通过下图，我们可以从管理视角直观的看到各种信息，这对于管理员来讲，省时省力，方便快捷。

图6

图6

意外收获：无缝迁移业务 Elasticsearch 上云

由于前期日志业务上云，受到非常好的反馈，多个业务线期望能够上云上服务，达到降本增效的目的。

支持双写

数据可以通过极限网关同时写入两个 ES 集群，能够保障数据完全一致，安全可靠。

无缝切换

切换很丝滑，影响非常小，能够让外界几乎感受不到服务波动。

图7

图7

通过使用极限网关，自建 ES 集群可以无缝的迁移上云，在整个迁移的过程中，两套集群通过网关进行了解耦，在迁移的过程中还能实现版本的无缝升级，极大降低了迁移成本，提高迁移效率，多次验证服务稳定可靠。

极限网关流量概览

这是其中一套极限网关的流量统计。用这部分数据进行巡检，一目了然，做到全局的掌控，提高感知力度。

图8

图8

极限网关使用总结

极限网关提供一系列高性能和高可靠性的网关服务。使用这样的服务给我们带来以下好处：

  1. 可观测性好：极限网关可以动态的对 Elasticsearch 运行过程中请求进行拦截和分析，通过指标和日志来了解集群运行状态，这些指标可以用于提升性能和业务优化。
  2. 增强安全性：包含先进的安全机制，如 basicauth、LDAP 等支持，保护用户数据不受未授权访问和各种网络威胁的侵害。
  3. 高稳定性：通过冗余设计和故障转移机制，极限网关能够确保网络服务的高可用性，即使在某些组件发生故障时也能保持服务不中断，单版本最长服务超过 15 个月。
  4. 易于管理：通过提供 INFINI Console 简洁直观的管理界面，让用户能够轻松配置和监控网络状态，提升管理效率。
  5. 客户支持：良好的客户服务支持可以帮助用户快速解决使用过程中遇到的问题，提供专业的技术指导。

综上所述，极限网关为用户提供了一个高速、安全、稳定且易于管理的 ES 网关，适合对网络性能有较高要求的个人和企业用户。

未来规划

第一阶段，完成了日志 ES 集群，所有集群的容器化改造，合并，成功的把成本降低了 60%以上。这期间积累了丰富容器化经验，为业务 ES 集群上容器做了良好的铺垫；成本优势和运维优势吸引越来越多的业务接入到容器化 ES 集群。

提升 ES 集群效能--新技术应用&&版本升级

  • 极限科技官方推荐的 Easysearch 在压缩率，查询速度等等方面有很多的优势，通过长时间的测试稳定性，新特性，对比云原生的 ES 集群，根据测试结果，给“客户”提供多种选择，这也是工作重点之一。
  • 我们当前使用的 ES 版本是 6.8，已经远远落后于官方版本，今年我们计划在选择合适的集群升级 ES 版本，拥抱更多官方提供的特性。

混合（多）云架构支持

随着越来越多的 ES 集群在机房的 K8S 集群部署，这里资源出现了紧张局面。 我们尝试在云上部署自建 ES 集群，弥补机房资源有限，无法大规模扩容，同时能够支持多活场景，满足更多客户的不同需求。混合云主要实现以下几种能力：

1、扩缩容：满足不同业务灵活适配

混合（多）云部署，可以让负载内部私有云 ES，同时部署到公有云，提升扩展 IT 基础设施不仅局限于 CPU、内存，还有存储。比如某一个业务要做活动，预估流量“大爆发”，需要提前准备大规模资源，在机房内根本来不及采购扩容支持，然而在公有云上就能很方便扩容、缩容。在云上搭建 ES 集群，设置满足需求的数量、容量、配置，配合极限网关路由策略，精准的把控流量流向。

图9

图9

2、灾备：紧急情况快速部署，恢复 ES 集群读写

当机房级别大规模故障，部分业务实现了多活，单一的机房故障不会影响其服务能力，而此时比如日志查看等仍有需求，为了满足这部分“客户”需求，可以在云上 K8S 集群，快速搭建 ES 集群，恢复日志读写功能。

图10

图10

参考文档：

作者：张华勋，前新浪 CDN 研发，工作主要涉及 Mysql、MongoDB、Redis、Elasticsearch、流量调度等组件和系统，以及运维自动化、平台化等工作。现就职于好未来。

关于好未来

好未来（NYSE：TAL）是一家以内容能力与科技能力为基础，以科教、科创、科普为战略方向，助力人的终身成长，并持续探索创新的科技公司。 好未来的前身学而思成立于 2003 年，2010 年在美国纽交所正式挂牌交易。好未来以“爱与科技助力终身成长”为使命，致力成为持续创新的组织。更多参见：https://www.100tal.com/

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.cn

联系我们

INFINI Labs 产品更新 | Easysearch 1.8.2 发布优化 CCR 性能

Easysearch • liaosy 发表了文章 • 0 个评论 • 3243 次浏览 • 2024-06-09 12:52 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Easysearch v1.8.2、Gateway、Console、Agent、Loadgen v1.26.0。本次各产品更新了很多亮点功能，如 Easysearch 优化 CCR 同步性能；Gateway 增加了 HTTP 请求动态域名路由功能，移除了安全相关的 Filter，进一步提升 Gateway 稳定性；Console 修复了多个已知问题，如当文档数过亿时单位换算错误，修复了因采集延迟导致指标图表显示异常，修复了多行查询中包含 SQL 查询异常等问题。欢迎大家下载体验。

以下是本次更新的详细说明。

INFINI Easysearch v1.8.2

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Bug fix

  • 修复 source_reuse 与 object 字段为 enable: false 时的冲突

Improvements

  • 升级部分依赖包版本，Commons-collections to 3.2.2, Snakeyaml to 2.0
  • 优化 CCR 同步性能及调整 CCR 全局配置参数
  • 优化插件配置命名，去除"plugins."
  • 优化配置文件目录获取命名

INFINI Console v1.26.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Bug fix

  • 修复监控数据布局
  • 修复命令存储权限
  • 修复多行请求包含 SQL 语法
  • 修复文档数过亿时换算错误
  • 修复导入低版本 v1.6.0 告警规则缺少字段问题
  • 修复当 buck_size 小于 60 秒时，因指标采集延迟导致指标显示异常问题

INFINI Gateway v1.26.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Improvements

  • feat: add wildcard_domain filter
  • chore: remove security filter and translog_viewer

INFINI Framework

INFINI Framework 是 INFINI Labs 各产品依赖的内部核心公共代码库。

Framework 本次更新如下：

Improvements

  • feat: support dynamic app setting
  • feat: add cluster settings query args
  • feat: add gateway config
  • feat: add http interceptor
  • feat: return host info in info api
  • feat: add util to convert string to float
  • feat: use common app setting api to instead of auth setting api
  • feat: crontab task support multi crontab expression
  • fix: skip submit empty bulk requests
  • feat: support ccr api
  • fix: get latest offset should compare segment first
  • fix: wrong use of zstd with vfs
  • fix: prevent close closed channel
  • fix: panic on error while saving keystore

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.cn

INFINI Labs 产品更新 | Console 1.24.0 操作日志审计功能发布

资讯动态 • liaosy 发表了文章 • 0 个评论 • 3903 次浏览 • 2024-04-28 12:51 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Console，Gateway 1.24.0。本次各产品更新了很多亮点功能，如 Console 增加操作日志审计功能，优化数据探索字段统计，修复 Gateway 增加认证后添加实例失败等问题。以下是本次更新的详细说明。

INFINI Console v1.24.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 用户操作审计日志功能
  • 新增告警规则克隆一键克隆功能，简化重复类型告警的创建

Bug fix

  • 修复普通用户权限 403 问题
  • 修复探针管理表格展开显示错位问题

Improvements

  • 优化开发工具集群选择控件显示位置
  • 优化数据探索查询控件显示宽度
  • 优化数据探索字段统计功能
  • 优化告警规则列表页搜索，支持远程搜索
  • Discover 左侧字段聚合支持开关控制拉取本地或远程统计值

INFINI Gateway v1.24.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Improvements

  • Refactoring http client tls config
  • Write field routing to bulk metadata when field _routing exists in scrolled doc

Bug fix

  • Fix(reshuffle filter): make sure queue config always have label type
  • Fix rotate config usage

INFINI Framework

Improvements

  • feat: allow to use default auth for agent’s auto enroll
  • refactor: refactor func GetFieldCaps
  • feat: register background job to clean up badger LSM tree
  • fix: skip load missing wal

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.cn

INFINI Labs 产品更新 | Easysearch 1.7.1发布，改进跨集群复制的数据加载等

Easysearch • liaosy 发表了文章 • 0 个评论 • 2691 次浏览 • 2024-03-04 15:25 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Console，Gateway，Agent 1.23.0 和 Easysearch 1.7.1。本次各产品更新了很多亮点功能，如 Console 优化实例管理中增加磁盘空闲空间显示，Easysearch 改进 HierarchyCircuitBreakerService 并添加断路器、改进跨集群复制的数据加载，增加对 source_reuse 索引的支持等。欢迎大家下载体验。

以下是本次更新的详细说明。

INFINI Easysearch v1.7.1

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Bug fix

  • 修复 source_reuse 与字段别名冲突
  • 改进 HierarchyCircuitBreakerService 并添加断路器
  • 修复 _meta 不为空且 启用 source_reuse 时的映射解析错误
  • 修复 source_reuse 下对多值还原不正确的问题

Improvements

  • 改进 HierarchyCircuitBreakerService 并添加断路器
  • 改进跨集群复制的数据加载，增加对 source_reuse 索引的支持

INFINI Console v1.23.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Bug fix

  • 修复数据迁移中数据分片范围因精度导致数据溢出显示为负数
  • 修复删除实例队列后消费的 Offset 未重置问题
  • 修复网友提出的各种问题，如集群设置默认打开节点、索引采集等

Improvements

  • 优化初始化配置向导，分步骤执行
  • 优化实例管理中增加磁盘空闲空间显示
  • 优化实例队列名称显示

INFINI Gateway v1.23.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Bug fix

  • 修复删除实例队列后消费的 Offset 未重置问题

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

如何防止 Elasticsearch 服务 OOM？

Easysearch • yangmf2040 发表了文章 • 0 个评论 • 3115 次浏览 • 2024-02-26 10:12 • 来自相关话题

Elasticsearch（简称：ES） 和传统关系型数据库有很多区别， 比如传统数据中普遍都有一个叫“最大连接数”的设置。目的是使数据库系统工作在可控的负载下，避免出现负载过高，资源耗尽，谁也无法登录的局面。

那 ES 在这方面有类似参数吗？答案是没有，这也是为何 ES 会被流量打爆的原因之一。

针对大并发访问 ES 服务，造成 ES 节点 OOM，服务中断的情况，极限科技旗下的 INFINI Gateway 产品（以下简称 “极限网关”）可从两个方面入手，保障 ES 服务的可用性。

  1. 限制最大并发访问连接数。
  2. 限制非重要索引的请求速度，保障重要业务索引的访问速度。

下面我们来详细聊聊。

架构图

所有访问 ES 的请求都发给网关，可部署多个网关。

限制最大连接数

在网关配置文件中，默认有最大并发连接数限制，默认最大 10000。

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 10000
    network:
      binding: $[[env.GW_BINDING]]
      # See `gateway.disable_reuse_port_by_default` for more information.
      reuse_port: true

使用压测程序测试，看看到达10000个连接后，能否限制新的连接。 超过的连接请求，被丢弃。更多信息参考官方文档。

限制索引写入速度

我们先看看不做限制的时候，测试环境的写入速度，在 9w - 15w docs/s 之间波动。虽然峰值很高，但不稳定。 接下来，我们通过网关把写入速度控制在最大 1w docs/s 。 对网关的配置文件 gateway.yml ，做以下修改。

env: # env 下添加
    THROTTLE_BULK_INDEXING_MAX_BYTES: 40485760 #40MB/s
    THROTTLE_BULK_INDEXING_MAX_REQUESTS: 10000 #10k docs/s
    THROTTLE_BULK_INDEXING_ACTION: retry #retry,drop
    THROTTLE_BULK_INDEXING_MAX_RETRY_TIMES: 10 #1000
    THROTTLE_BULK_INDEXING_RETRY_DELAY_IN_MS: 100 #10

router: # route 部分修改 flow
  - name: my_router
    default_flow: default_flow
    tracing_flow: logging_flow
    rules:
      - method:
          - "*"
        pattern:
          - "/_bulk"
          - "/{any_index}/_bulk"
        flow:
          - write_flow

flow: #flow 部分增加下面两段    
  - name: write_flow
    filter:
      - flow:
          flows:
            - bulking_indexing_limit
      - elasticsearch:
          elasticsearch: prod
          max_connection_per_node: 1000
  - name: bulking_indexing_limit
    filter:
      - bulk_request_throttle:
          indices:
            "test-index":
              max_bytes: $[[env.THROTTLE_BULK_INDEXING_MAX_BYTES]]
              max_requests: $[[env.THROTTLE_BULK_INDEXING_MAX_REQUESTS]]
              action: $[[env.THROTTLE_BULK_INDEXING_ACTION]]
              retry_delay_in_ms: $[[env.THROTTLE_BULK_INDEXING_RETRY_DELAY_IN_MS]]
              max_retry_times: $[[env.THROTTLE_BULK_INDEXING_MAX_RETRY_TIMES]]
              message: "bulk writing too fast" #触发限流告警message自定义
              log_warn_message: true

再次压测，test-index 索引写入速度被限制在了 1w docs/s 。

限制多个索引写入速度

上面的配置是针对 test-index 索引的写入速度控制。如果想添加其他的索引，新增一段配置即可。 比如，我允许 abc 索引写入达到 2w docs/s，test-index 索引最多不超过 1w docs/s ，可配置如下。

  - name: bulking_indexing_limit
    filter:
      - bulk_request_throttle:
          indices:
            "abc":
              max_requests: 20000
              action: drop
              message: "abc doc写入超阈值" #触发限流告警message自定义
              log_warn_message: true
            "test-index":
              max_bytes: $[[env.THROTTLE_BULK_INDEXING_MAX_BYTES]]
              max_requests: $[[env.THROTTLE_BULK_INDEXING_MAX_REQUESTS]]
              action: $[[env.THROTTLE_BULK_INDEXING_ACTION]]
              retry_delay_in_ms: $[[env.THROTTLE_BULK_INDEXING_RETRY_DELAY_IN_MS]]
              max_retry_times: $[[env.THROTTLE_BULK_INDEXING_MAX_RETRY_TIMES]]
              message: "bulk writing too fast" #触发限流告警message自定义
              log_warn_message: true

限速效果如下

限制读请求速度

我们先看看不做限制的时候，测试环境的读取速度，7w qps 。 接下来我们通过网关把读取速度控制在最大 1w qps 。 继续对网关的配置文件 gateway.yml 做以下修改。

  - name: default_flow
    filter:
      - request_path_limiter:
          message: "Hey, You just reached our request limit!"                                      rules:      
            - pattern: "/(?P<index_name>abc)/_search"                          
              max_qps: 10000
              group: index_name                                    
      - elasticsearch: 
          elasticsearch: prod                            
          max_connection_per_node: 1000

再次进行测试，读取速度被限制在了 1w qps 。

限制多个索引读取速度

上面的配置是针对 abc 索引的写入速度控制。如果想添加其他的索引，新增一段配置即可。 比如，我允许 abc 索引读取达到 1w qps，test-index 索引最多不超过 2w qps ，可配置如下。

  - name: default_flow
    filter:
      - request_path_limiter:
          message: "Hey, You just reached our request limit!"
          rules:
            - pattern: "/(?P<index_name>abc)/_search"
              max_qps: 10000
              group: index_name
            - pattern: "/(?P<index_name>test-index)/_search"
              max_qps: 20000
              group: index_name
      - elasticsearch:
          elasticsearch: prod
          max_connection_per_node: 1000

多个网关限速

限速是每个网关自身的控制，如果有多个网关，那么后端 ES 集群收到的请求数等于多个网关限速的总和。 本次介绍就到这里了。相信大家在使用 ES 的过程中也遇到过各种各样的问题。欢迎大家来我们这个平台分享自己的问题、解决方案等。如有任何问题，请随时联系我，期待与您交流！

INFINI Labs 产品更新 | Console 数据迁移支持 Percentiles 均匀分区

资讯动态 • liaosy 发表了文章 • 0 个评论 • 3807 次浏览 • 2023-12-30 16:25 • 来自相关话题

release

INFINI Labs 产品又更新啦~，包括 Console v1.14.0，Gateway 1.21.0。其中 Console 数据迁移支持 Percentiles 均匀分区，修复已知 Bug 等。以下是本次更新的详细说明。

INFINI Console v1.14.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 数据迁移支持 decimal 类型分区设置
  • 数据迁移支持 number 类型均匀分区

Bug fix

  • 修复告警恢复后，新周期内没有发送告警通知消息的问题

INFINI Gateway v1.21.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Bug fix

  • fix log error when temp file was missing

INFINI Framework

INFINI Framework 是 INFINI Labs 各产品依赖的内部核心公共代码库。

Framework 本次更新如下：

  • feat: support even partition
  • feat: support nested object
  • fix log error when temp file was missing

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

INFINI Labs 产品更新 | 修复 Easysearch 跨集群复制索引同步问题，Gateway 内存异常增长等问题

Easysearch • liaosy 发表了文章 • 0 个评论 • 3021 次浏览 • 2023-12-01 18:45 • 来自相关话题

release

INFINI Labs 产品又更新啦~，本次更新主要对 Easysearch、Gateway、Console、Agent 等产品功能进行优化和相关 Bug 修复，解决了内存异常增长等问题，以下是详细说明。

INFINI Easysearch v1.6.2

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Bug fix

  • 修复跨集群复制（CCR）不能对自动滚动生成的索引进行同步的问题

Improvements

  • 优化初始化脚本，增加-s/-slient 参数，自动安装。
  • 新增含 jdk/plugins 的 bundle 安装包

INFINI Gateway v1.20.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Bug fix

  • 修复由 Framework Bug 造成连接数不释放、内存异常增长的问题

Improvements

  • 增加配置，允许设置 fasthttp client 相关参数

INFINI Console v1.12.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Bug fix

  • 修复数据探索 multi fields 字段计算 top values 报错的问题
  • 修复由 Framework Bug 造成连接数不释放、内存异常增长的问题
  • 修复内网模式下静态资源远程加载的问题
  • 修复数据看板数据源配置校验异常的问题

Improvements

  • 优化数据探索计算 top values，使用先采样后，后取 top values
  • 可通过配置参数 http_client.read_buffer_size 设置读取缓存大小，解决开发工具执行命令时，默认缓存太小的问题

INFINI Agent v0.7.1

INFINI Agent 是 INFINI Console 的一个可选探针组件，负责采集和上传集群指标和日志等信息，并可通过 Console 管理。Agent 支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。

Agent 本次更新如下：

Features

  • 添加 http processor

Bug fix

  • 修复由 Framework Bug 造成连接数不释放、内存异常增长的问题

Improvements

  • 进一步优化内存占用，降到 50M 以下

INFINI Framework

INFINI Framework 是 INFINI Labs 各产品依赖的内部核心公共代码库。

Framework 本次更新如下：

  • fix: fix the issue of disk queue was blocked
  • chore: checkout specify branch before pull

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

INFINI Labs 产品更新 | 发布 Easysearch Java 客户端，Console 支持 SQL 查询等功能

Easysearch • liaosy 发表了文章 • 0 个评论 • 3711 次浏览 • 2023-11-17 18:56 • 来自相关话题

release

INFINI Labs 产品又更新啦~，本次更新概要如下：发布 Easysearch-client Java 客户端，开发者通过 client 与 Easysearch 集群的交互变得更加简洁和直观；Console 开发工具新增 SQL 特性，支持 SELECT 查询等语法高亮和自动提示等；Gateway 的系统 API 添加了基于基本身份验证的安全功能。

以下是本次更新的详细说明。

INFINI Easysearch-client v1.0.1

正式发布 Easysearch Java 客户端。

这一里程碑式的更新为开发人员带来了前所未有的便利性，使得与 Easysearch 集群的交互变得更加简洁和直观。现在，通过 Easysearch-client 客户端，开发者可以直接使用 Java 方法和数据结构来进行交互，而不再需要依赖于传统的 HTTP 方法和 JSON。这一变化大大简化了操作流程，使得数据管理和索引更加高效。高级客户端的功能范围包括处理数据操作，管理集群，包括查看和维护集群的健康状态，并对 Security 模块全面兼容。它提供了一系列 API，用于管理角色、用户、权限、角色映射和账户。这意味着安全性和访问控制现在可以更加细粒度地管理，确保了数据的安全性和合规性。

使用说明参见：快速开始

INFINI Console v1.11.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 开发工具 SQL 查询支持
    • 支持 SELECT 查询及语法高亮
    • 支持索引和字段自动提示
    • 支持 FROM 前置语法

Bug fix

  • 修复平台概览集群指标为空的问题

Improvements

  • LDAP 支持从 DN 中解析 OU 属性
  • 集群动态优化显示，新增节点名称和索引名称的聚合统计过滤

INFINI Gateway v1.19.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Features

  • 添加 http 处理器
  • 在 API 模块中添加基于基本身份验证的安全性
  • 允许将自身注册到配置管理器
  • 允许在配置错误时触发 panic

Bug fix

  • 修复 rewrite_to_bulk 在较新版本中缺少 _type 的问题
  • 修复 rewrite_to_bulk，支持无索引文档操作

Improvements

  • 更新测试，断言解析结果

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

INFINI Labs 产品更新 | Gateway 支持基于 Kafka 的复制能力，发布 Helm Charts 部署方式

资讯动态 • liaosy 发表了文章 • 0 个评论 • 2858 次浏览 • 2023-09-22 16:16 • 来自相关话题

release

INFINI Labs 产品又更新啦~。本次更新概要如下：Easysearch 新增了索引字段相关统计 API，优化了 source_reuse 提升压缩效率；Gateway 新增诸多新特性，如：支持基于 Kafka 的复制能力，添加可插拔的分布式锁实现，新增 CPU 资源限制等功能；Console 本次主要优化了数据迁移功能，迁移任务详情页新增了若干指标图和日志查看等功能。

欢迎大家下载使用和反馈。

INFINI Helm Charts v0.1.0

INFINI Helm Charts 是一组 Kubernetes 部署包管理工具。基于 Helm Charts，我们将 INFINI Labs 旗下相关产品预先配置好程序资源包，大大简化了部署流程。Github 仓库地址：https://github.com/infinilabs/helm-charts。

Helm Charts 本次更新如下：

Features

  • 添加 Console Chart
  • 添加 Easysearch Chart，支持单节点以及多节点（节点角色可配置）部署

部署视频演示：

博客文章：

INFINI Easysearch v1.6.0

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Features

  • 新增 _field_usage_stats API，统计索引每个字段的访问次数
  • 新增 _disk_usage API，可以分析指定索引每个字段的磁盘占用大小
  • 增加 flattened 类型，将 JSON 对象作为字符串处理，可以减少嵌套 JSON 型的文档的大小

Improvements

  • source_reuse 增加对 _source 中数字类型的值进行复用压缩，可进一步降低 _source 磁盘占用
  • 改进 source_reuse 筛选字段的逻辑

INFINI Gateway v1.18.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Breaking changes

  • 彻底移除了 request_body_truncate 和 response_body_truncate 过滤器

Features

  • 支持基于 Kafka 的复制能力
  • 在请求上下文中添加 _util.generate_uuid
  • 在请求上下文中添加 _util.increment_id.BUCKET_NAME
  • 在 Pipeline 配置中添加 singleton，防止多个 Pipeline 同时运行
  • 添加可插拔的分布式锁实现
  • 添加通用应用程序的 preference 配置
  • 泛化队列抽象，重构磁盘队列，完善 Kafka 实现
  • 添加 merge_to_bulk 处理器， 废弃 indexing_merge 处理器
  • 添加 flow_replay 处理器，废弃 flow_runner 处理器
  • 为复制场景添加 replication_correlation
  • 添加 hash_mod 过滤器
  • 在 bulk_response_process 过滤器中添加新参数
  • 添加 request_reshuffle 过滤器
  • 添加资源限制，允许设置最大 CPU 数或绑定亲和性
  • 支持模板中的嵌套变量
  • 添加 rewrite_to_bulk 过滤器

Bug fix

  • 修复了 Pipeline 中重试延迟未生效的问题
  • 修复了模板中不支持数字的问题
  • 修复了队列选择器通过标签的问题，如果指定了多个标签，它们都应该一起匹配

Improvements

  • 将所有模块名称转换为小写
  • 在启动期间预取 Elasticsearch 元数据
  • 添加应用程序范围的关闭信号
  • 重构队列 API，支持 Kafka 管理
  • 在 Badger 模块中添加 enabled
  • 允许使用优先级注册模块/插件
  • 统一队列的使用和初始化
  • 优化 bulk_reshuffle 过滤器的性能，添加响应头 X-Bulk-Reshuffled
  • 支持在 queue 过滤器中使用变量，允许输出最后生成的消息偏移量

INFINI Console v1.8.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 数据迁移任务支持自定义名称和添加标签
  • 数据迁移任务详情页新增若干指标
  • 数据迁移任务详情页新增查看日志

Improvements

  • 数据迁移 UI 优化
  • 优化监控报表、数据看板、数据探索的时间控件 UI

数据迁移列表页

数据迁移详情页

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.com/invite/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

联系我们

关于极限科技（INFINI Labs）

INFINI Labs

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

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

官网：https://www.infinilabs.com

INFINI Labs 产品更新 | Easysearch 新增分词插件、Gateway 支持邮件发送等功能

资讯动态 • liaosy 发表了文章 • 0 个评论 • 2567 次浏览 • 2023-07-21 17:28 • 来自相关话题

INFINI Labs 产品又更新啦~，本次更新概要如下：Easysearch 新增了分词插件、优化了生命周期管理功能等；Gateway 新增 smtp 过滤器来支持邮件的发送，支持自动跳过因为异常关闭而损坏的磁盘队列文件等；Console 新增熔断器监控指标、新增矩形树图(Treemap)、优化了探针 Agent 指标采集和集群自动关联操作等。欢迎大家下载体验。

INFINI Easysearch v1.4.0

INFINI Easysearch 是一个分布式的近实时搜索与分析引擎，核心引擎基于开源的 Apache Lucene。Easysearch 衍生自基于开源协议 Apache 2.0 的 Elasticsearch 7.10 版本。Easysearch 的目标是提供一个轻量级的 Elasticsearch 可替代版本，并继续完善和支持更多的企业级功能。

Easysearch 本次更新如下：

Features

  • 索引生命周期管理增加 wait_for_snapshot 操作，在删除索引之前，等待执行指定的快照管理策略，这样可以确保已删除索引的快照可用
  • 增加 analysis-hanlp 分词插件
  • 增加 jieba 分词插件

Bug fix

  • 修复启用 index.source_reuse 时，对复杂多层 json 的 source 字段 解析不正确的 Bug

Improvements

  • 更新索引生命周期管理 api 文档，增加策略应用和更新说明，增加 wait_for_snapshot 说明
  • 执行 initialize.sh 命令时增加初始化确认提示，是否将 admin 密码记录日志

INFINI Gateway v1.17.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Features

  • 新增 consumer Processor 来标准化订阅消息队列
  • 新增 smtp 过滤器来支持邮件的发送

Bug fix

  • 支持自动跳过因为异常关闭而损坏的磁盘队列文件

INFINI Console v1.4.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

  • 新增熔断器监控指标
  • 网关队列管理支持多选删除消费者
  • 数据看板新增组件矩形树图

Bug fix

  • 修复开发工具智能提示兼容性问题
  • 修复探针列表状态显示异常的问题
  • 修复探针列表分页不生效的问题
  • 修复数据看板 统计函数显示不对的问题
  • 修复探针下发采集指标配置重复的问题
  • 修复数据看板设置过滤条件不生效的问题
  • 修复主机列表探针状态不对的问题
  • 修复网关管理删除队列不成功的问题
  • 修复数据探索当前集群没索引时跳转的问题
  • 修复数据看板编辑状态下点击事件的问题

Improvements

  • 探针进程关联支持通过选择集群自动关联，简化操作
  • 探针列表支持排序
  • 探针支持向上滚动查看节点日志
  • 采集监控指标添加 cluster_uuid 信息
  • 数据看板支持配置指标排序

INFINI Agent v0.6.0

INFINI Agent 是 INFINI Console 的一个可选探针组件，负责采集和上传集群指标和日志等信息，并可通过 Console 管理。Agent 支持主流操作系统和平台，安装包轻量且无任何外部依赖，可以快速方便地安装。

Agent 本次更新如下：

Features

  • 采集监控指标添加 cluster_uuid 信息

Improvements

  • 修复发现节点进程信息时获取不到最新集群配置的问题

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

或者通过 Discord 渠道加入聊天室：https://discord.com/invite/tnZ8S5vQ

欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群讨论，或者扫码加入我们的知识星球一起学习交流。

联系我们

关于极限科技（INFINI Labs）

关于极限科技

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

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

官网：https://www.infinilabs.com

用极限网关实现ES容灾，简单！

Elasticsearch • yangmf2040 发表了文章 • 0 个评论 • 3508 次浏览 • 2023-07-20 10:33 • 来自相关话题

身为 IT 人士，大伙身边的各种系统肯定不少吧。系统虽多，但最最最重要的那套、那几套，大伙肯定是捧在手心，关怀备至。如此重要的系统，万一发生故障了且短期无法恢复，该如何保障业务持续运行？ 有过这方面思考或经验的同学，肯定脱口而出--切灾备啊。 是的，接下来我来介绍下我们的 ES 灾备方案。当然如果你有更好的，请使用各种可用的渠道联系我们。

总体设计

通过极限网关将应用对主集群的写操作，复制到灾备集群。应用发送的读请求则直接转发到主集群，并将响应结果转发给应用。应用对网关无感知，访问方式与访问 ES 集群一样。

方案优势

  • 轻量级

极限网关使用 Golang 编写，安装包很小，只有 10MB 左右，没有任何外部环境依赖，部署安装都非常简单，只需要下载对应平台的二进制可执行文件，启动网关程序的二进制程序文件执行即可。

  • 跨版本支持

极限网关针对不同的 Elasticsearch 版本做了兼容和针对性处理，能够让业务代码无缝的进行适配，后端 Elasticsearch 集群版本升级能够做到无缝过渡，降低版本升级和数据迁移的复杂度。

  • 高可用

极限网关内置多种高可用解决方案，前端请求入口支持基于虚拟 IP 的双机热备，后端集群支持集群拓扑的自动感知，节点上下线能自动发现，自动处理后端故障，自动进行请求的重试和迁移。

  • 灵活性

主备集群都是可读可写，切换迅速，只需切换网关到另一套配置即可。回切灵活，恢复使用原配置即可。

架构图

网关程序部署

下载

根据操作系统和平台选择下面相应的安装包： 解压到指定目录：

mkdir gateway
tar -zxf xxx.gz -C gateway

修改网关配置

在此 下载 网关配置，默认网关会加载配置文件 gateway.yml ，如果要指定其他配置文件使用 -config 选项指定。 网关配置文件内容较多，下面展示必要部分。

  #primary
  PRIMARY_ENDPOINT: http://192.168.56.3:7171
  PRIMARY_USERNAME: elastic
  PRIMARY_PASSWORD: password
  PRIMARY_MAX_QPS_PER_NODE: 10000
  PRIMARY_MAX_BYTES_PER_NODE: 104857600 #100MB/s
  PRIMARY_MAX_CONNECTION_PER_NODE: 200
  PRIMARY_DISCOVERY_ENABLED: false
  PRIMARY_DISCOVERY_REFRESH_ENABLED: false
  #backup
  BACKUP_ENDPOINT: http://192.168.56.3:9200
  BACKUP_USERNAME: admin
  BACKUP_PASSWORD: admin
  BACKUP_MAX_QPS_PER_NODE: 10000
  BACKUP_MAX_BYTES_PER_NODE: 104857600 #100MB/s
  BACKUP_MAX_CONNECTION_PER_NODE: 200
  BACKUP_DISCOVERY_ENABLED: false
  BACKUP_DISCOVERY_REFRESH_ENABLED: false

PRIMARY_ENDPOINT：配置主集群地址和端口
PRIMARY_USERNAME、PRIMARY_PASSWORD: 访问主集群的用户信息
BACKUP_ENDPOINT：配置备集群地址和端口
BACKUP_USERNAME、BACKUP_PASSWORD: 访问备集群的用户信息

运行网关

前台运行 直接运行网关程序即可启动极限网关了，如下：

./gateway-linux-amd64

后台运行

./gateway-linux-amd64 -service install
Success
./gateway-linux-amd64 -service start
Success

卸载服务

./gateway-linux-amd64 -service stop
Success
./gateway-linux-amd64 -service uninstall
Success

灾备功能测试

在灾备场景下，为保证数据一致性，对集群的访问操作都通过网关进行。注意只有 bulk API 的操作才会被复制到备集群。 在此次测试中，网关灾备配置功能为：

  • 主备集群正常时

读写请求正常执行； 写请求被记录到队列，备集群实时消费队列数据。

  • 当主集群故障时

写入请求报错，主备集群都不写入数据； 查询请求转到备集群执行，并返回结果给客户端。

  • 当备集群故障时

读写请求都正常执行； 写操作记录到磁盘队列，待备集群恢复后，自动消费队列数据直到两个集群一致。

主备集群正常时写入、查询测试

写入数据

# 通过网关写入数据
curl -X POST "localhost:18000/_bulk?pretty" -H 'Content-Type: application/json' -uelastic:password -d'
{ "index" : { "_index" : "test", "_id" : "1" } }
{ "field1" : "value1" }
{ "create" : { "_index" : "test", "_id" : "2" } }
{ "field2" : "value2" }
'

查询数据

# 查询主集群
curl 192.168.56.3:7171/test/_search?pretty -uelastic:password

# 查询备集群
curl 192.168.56.3:9200/test/_search?pretty -uadmin:admin

# 查询网关，网关转发给主集群执行
curl 192.168.56.3:18000/test/_search?pretty -uelastic:password

主备集群都已写入数据，且数据一致。通过网关查询，也正常返回。

删除和更新文档

# 通过网关删除和更新文档
curl -X POST "192.168.56.3:18000/_bulk?pretty" -H 'Content-Type: application/json' -uelastic:password -d'
{ "delete" : { "_index" : "test", "_id" : "1" } }
{ "update" : {"_id" : "2", "_index" : "test"} }
{ "doc" : {"field2" : "value2-updated"} }
'

查询数据
# 查询主集群
curl 192.168.56.3:7171/test/_search?pretty -uelastic:password

# 查询备集群
curl 192.168.56.3:9200/test/_search?pretty -uadmin:admin

两个集群都已执行删除和更新操作，数据一致。

主集群故障时写入、查询测试

为模拟主集群故障，直接关闭主集群。

写入数据

# 通过网关写入数据
curl -X POST "192.168.56.3:18000/_bulk?pretty" -H 'Content-Type: application/json' -uelastic:password -d'
{ "index" : { "_index" : "test", "_id" : "3" } }
{ "field3" : "value3" }
{ "create" : { "_index" : "test", "_id" : "4" } }
{ "field4" : "value4" }
'

写入数据报错

查询数据

# 通过网关查询，因为主集群不可用，网关将查询转发到备集群执行
curl 192.168.56.3:18000/test/_search?pretty -uelastic:password

正常查询到数据，说明请求被转发到了备集群执行。

备集群故障时写入、查询测试

为模拟备集群故障，直接关闭备集群。

写入数据

# 通过网关写入数据
curl -X POST "192.168.56.3:18000/_bulk?pretty" -H 'Content-Type: application/json' -uelastic:password -d'
{ "index" : { "_index" : "test", "_id" : "5" } }
{ "field5" : "value5" }
{ "create" : { "_index" : "test", "_id" : "6" } }
{ "field6" : "value6" }
'

数据正常写入。

查询数据

# 通过网关查询
curl 192.168.56.3:18000/test/_search?pretty -uelastic:password

查询成功返回。主集群成功写入了两条新数据。同时此数据会被记录到备集群的队列中，待备集群恢复后，会消费此队列追数据。

恢复备集群

启动备集群。

查询数据

等待片刻或通过 INFINI Console 确定网关队列消费完毕后，查询备集群的数据。 (生产和消费 offset 相同，说明消费完毕。) 

# 查询备集群的数据
curl 192.168.56.3:9200/test/_search?pretty -uadmin:admin

备集群启动后自动消费队列数据，消费完后备集群数据达到与主集群数据一致。

灾备切换

测试了这么多，终于到切换的时刻了。切换前我们判断下主系统是否短期无法修复。

如果我们判断主用系统无法短时间恢复，要执行切换。非常简单，我们直接将配置文件中定义的主备集群互换，然后重启网关程序就行了。但我们推荐在相同主机上另部署一套网关程序--网关B，先前那套用网关A指代。网关B中的配置文件把原备集群定义为主集群，原主集群定义为备集群。若要执行切换，我们先停止网关A，然后启动网关B，此时应用连接到网关(端口不变)，就把原备系统当作主系统使用，把原主系统当作备系统，也就完成了主备系统的切换。

灾备回切

当原主集群修复后，正常启动，就会从消费队列追写修复期间产生数据直到主备数据一致，同样我们可通过 INFINI Console 查看消费的进度。如果大家还是担心数据的一致性，INFINI Console 还能帮大家做[校验数据]()任务，做到数据完全一致后(文档数量及文档内容一致)，才进行回切。

回切也非常简单，停止网关B，启动网关A即可。

网关高可用

网关自带浮动 IP 模块，可进行双机热备。客户端通过 VIP 连接网关，网关出现故障时，VIP 漂移到备网关。 视频教程戳这里。

这样的优点是简单，不足是只有一个网关在线提供服务。如果想多个网关在线提供服务，则需搭配分布式消息系统一起工作，架构如下。

前端通过负载均衡将流量分散到多个在线网关，网关将消息存入分布式消息系统。此时，网关可看作无状态应用，可根据需要扩缩规模。

以上就是我介绍的ES灾备方案，是不是相当灵活了。有问题还是那句话 Call me 。

关于极限网关

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway，可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

官网文档：https://www.infinilabs.com/docs/latest/gateway

更多...
Copyright © 2025 · CC BY-NC-SA 3.0

本站服务器及带宽由 提供赞助