OpenObserve API开发指南：从数据 Ingestion 到查询的完整接口

苏舰孝Noel

1273人浏览 · 2025-11-08 03:17:03

苏舰孝Noel · 2025-11-08 03:17:03 发布

OpenObserve API开发指南：从数据 Ingestion 到查询的完整接口

【免费下载链接】openobserve 🚀 10x easier, 🚀 140x lower storage cost, 🚀 high performance, 🚀 petabyte scale - Elasticsearch/Splunk/Datadog alternative for 🚀 (logs, metrics, traces, RUM, Error tracking, Session replay). 项目地址: https://gitcode.com/GitHub_Trending/op/openobserve

OpenObserve作为新一代可观测性平台，提供了高效的数据摄入（Ingestion）与查询能力。本文将详细介绍其API接口设计，帮助开发者快速实现从数据采集到检索分析的全流程集成。作为Elasticsearch/Splunk的替代方案，OpenObserve通过优化的存储结构和查询引擎，实现了140倍存储成本降低和更高的性能表现，特别适合日志、指标、追踪等观测数据的处理。

核心架构概览

OpenObserve采用模块化设计，核心功能通过Ingester（摄入器）和Searcher（查询器）两大组件实现。Ingester负责数据接收、处理与持久化，Searcher则提供SQL和PromQL查询能力。系统架构如图所示：

核心模块路径：

摄入服务实现：src/service/ingestion/mod.rs
查询服务实现：src/service/search/mod.rs

数据摄入API详解

1. 认证与基础配置

所有API请求需包含认证信息，通过HTTP头传递：

Authorization: Basic <base64_encoded_credentials>

基础URL结构：http://<host>:5080/api/<org_id>/<stream_type>/<stream_name>

其中：

org_id: 组织ID，默认使用"default"
stream_type: 数据类型，支持logs/metrics/traces
stream_name: 数据流名称，需提前创建

2. 日志数据摄入

支持JSON格式批量写入，通过POST请求实现高效数据摄入：

curl -X POST "http://localhost:5080/api/default/logs/app-logs" \
  -H "Authorization: Basic YWRtaW46YWRtaW4=" \
  -H "Content-Type: application/json" \
  -d '[
    {"timestamp": 1678900000000, "level": "info", "message": "User login", "user_id": "123"},
    {"timestamp": 1678900001000, "level": "error", "message": "Payment failed", "user_id": "456"}
  ]'

成功响应（200 OK）：

{
  "code": 200,
  "status": "success",
  "message": "Ingested 2 records",
  "records": 2,
  "size": 256
}

数据摄入处理逻辑在src/service/ingestion/mod.rs中实现，包含数据验证、分区键生成和Parquet文件写入等核心步骤。

3. OpenTelemetry协议支持

原生支持OTLP协议，可直接接收OpenTelemetry采集的追踪数据：

curl -X POST "http://localhost:5080/v1/traces" \
  -H "Content-Type: application/json" \
  -H "Authorization: Basic YWRtaW46YWRtaW4=" \
  -d @trace_payload.json

追踪数据可视化界面如图所示：

数据处理管道API

OpenObserve提供实时数据处理管道，支持数据清洗、转换和富化。通过API可创建和管理处理规则：

创建数据处理管道

POST /api/default/pipelines
Content-Type: application/json

{
  "name": "log_enrichment",
  "description": "Add metadata to logs",
  "stream": "app-logs",
  "stream_type": "logs",
  "functions": [
    {
      "id": "1",
      "type": "vrl",
      "source": ".metadata.host = downcase(.metadata.host)"
    }
  ]
}

VRL（Vector Remap Language）是OpenObserve的数据转换语言，编译和执行逻辑在src/service/ingestion/mod.rs的compile_vrl_function和apply_vrl_fn函数中实现。

查询API使用指南

1. SQL查询接口

通过POST请求执行SQL查询，支持复杂的聚合分析：

curl -X POST "http://localhost:5080/api/default/_search" \
  -H "Authorization: Basic YWRtaW46YWRtaW4=" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "SELECT level, count(*) as cnt FROM logs.app-logs WHERE timestamp > now() - 3600 GROUP BY level",
    "format": "json"
  }'

查询结果示例：

{
  "took": 120,
  "hits": [
    {"level": "info", "cnt": 1530},
    {"level": "error", "cnt": 42}
  ],
  "total": 2,
  "scan_size": 102400
}

2. 高级查询功能

支持时间序列分析、直方图和地理空间查询等高级功能。例如，生成错误率趋势图：

SELECT 
  histogram(timestamp, INTERVAL '5 minutes') as time_bucket,
  countIf(level = 'error') / count(*) as error_rate
FROM logs.app-logs
WHERE timestamp > now() - INTERVAL '1 day'
GROUP BY time_bucket
ORDER BY time_bucket

查询结果可直接用于构建监控仪表盘：

批量操作与性能优化

1. 批量摄入最佳实践

使用gzip压缩减少网络传输：Content-Encoding: gzip
批量大小控制在1MB-10MB之间
设置适当的超时时间：X-Request-Timeout: 30000

2. 查询性能优化

限制时间范围：始终在WHERE子句中指定timestamp范围
使用分区键过滤：利用get_stream_partition_keys函数获取的分区信息
合理使用索引：在src/service/search/mod.rs中实现的索引优化逻辑

完整工作流示例

以下是一个从数据采集到查询分析的完整Python示例：

import requests
import json
import time

BASE_URL = "http://localhost:5080/api/default"
AUTH = ("admin", "Complexpass#123")

# 1. 摄入示例日志数据
def ingest_logs():
    logs = [
        {
            "timestamp": int(time.time() * 1000),
            "level": "info",
            "message": "User login",
            "user_id": f"user_{i}",
            "metadata": {"host": f"server_{i%3}"}
        }
        for i in range(100)
    ]
    
    response = requests.post(
        f"{BASE_URL}/logs/app-logs",
        auth=AUTH,
        headers={"Content-Type": "application/json"},
        data=json.dumps(logs)
    )
    return response.json()

# 2. 执行查询分析
def query_logs():
    query = {
        "query": """
            SELECT 
                metadata.host, 
                count(*) as login_count 
            FROM logs.app-logs 
            WHERE timestamp > now() - 3600 
            GROUP BY metadata.host
        """,
        "format": "json"
    }
    
    response = requests.post(
        f"{BASE_URL}/_search",
        auth=AUTH,
        headers={"Content-Type": "application/json"},
        data=json.dumps(query)
    )
    return response.json()

# 执行工作流
if __name__ == "__main__":
    ingest_result = ingest_logs()
    print(f"Ingested: {ingest_result['records']} records")
    
    query_result = query_logs()
    print("Query results:")
    for hit in query_result['hits']:
        print(f"Host: {hit['host']}, Logins: {hit['login_count']}")

监控与告警集成

OpenObserve提供完整的告警API，可基于查询结果配置告警规则：

POST /api/default/alerts
Content-Type: application/json

{
  "name": "high_error_rate",
  "description": "Alert when error rate exceeds 5%",
  "stream": "app-logs",
  "stream_type": "logs",
  "query": "SELECT countIf(level='error')/count(*) as error_rate FROM logs.app-logs WHERE timestamp > now() - 60",
  "condition": "error_rate > 0.05",
  "frequency": 60,
  "destination": {
    "type": "slack",
    "url": "https://hooks.slack.com/services/XXX"
  }
}

告警管理界面：