OpenObserve 是一套基于 Rust 语言的开源可观测数据融合处理平台。通过不同的 api 接口，支持 metrics、log、trace 等数据的持久化。本文尝试探究一下 OpenObserve 的运行原理。
1、安装
官方给了 2 种典型的安装方式，包括单机模式和 HA 高可用模式。详细可参考https://openobserve.ai/docs/quickstart/
1.1 单机模式
运行很简单，携带用户名和密码，启动即可。
ZO_ROOT_USER_EMAIL="root@example.com" ZO_ROOT_USER_PASSWORD="Complexpass#123" ./openobserve
复制代码
1.2 HA 模式
与 Cortex/Thanos 等时序数据持久化解决方案类似，OpenObserve 的高可用模式也包括多个功能组件：ingester、compactor、querier、alert。特别之处在于，etcd 存放 series 的 schema 信息，S3 存放数据的 parquet 文件。在 k8s 集群中，参考官方指引部署即可：
helm repo add openobserve https://charts.openobserve.aihelm repo update
kubectl create ns openobserve
helm --namespace openobserve -f values.yaml install zo1 openobserve/openobserve
#暴露前端页面kubectl --namespace openobserve port-forward svc/zo1-openobserve-router 5080:5080
复制代码
本例使用到的 values.yaml 除了使用官方文件进行修改，也可以参考本文相关的https://github.com/grafanafans/club/tree/master/OpenObserve链接。
helm --namespace openobserve -f values.yaml install zo1 openobserve/openobserve
replicaCount: Not a table.NAME: zo1LAST DEPLOYED: Wed Sep 27 12:22:05 2023NAMESPACE: openobserveSTATUS: deployedREVISION: 1TEST SUITE: None
Forwarding from 127.0.0.1:5080 -> 5080Forwarding from [::1]:5080 -> 5080
复制代码
2、数据注入
2.1 metrics 写入
接口调用
POST /api/{organization}/prometheus/api/v1/write
复制代码
数据写入支持 OTEL collector 和 Prometheus 等方式。本文以 Prometheus 数据导入为例。参考 Prometheus 的 remote write 配置如下：
remote_write: - url: "http://127.0.0.1:5080/api/org_name/prometheus/api/v1/write" basic_auth: username: root@example.com password: Complexpass#123
复制代码
2.2 log 写入
接口调用
POST /api/{organization}/{stream}/_json
复制代码
在 vector、Filebeat、Syslog、Curl 等多种方式下快捷导入日志信息。
2.3 trace 写入
接口调用
POST /api/{organization}/traces
复制代码
支持 opentelemetry SDK 的原生格式，导入 trace 数据，包括但不限于 Go、Python、Java 等语言。
3、数据流转
在 twitter 上咨询 OpenObserve 时，他们分享了两点信息：
1. OpenObserve stores files temporarily as json before being converted to parquet files. All series pertaining to a single metric are stored together.2. sled/etcd eventually stores metadata in physical files, but OpenObserve never stores data directly to files. Only through sled/etcd.
复制代码
所以，OpenObserve 不再像 Prometheus 一样，去维护 index/tsdb 信息，而是交由第三方工具，来进行 meta 信息，以及落盘文件的格式管理。下面我们以 metrics 的写入和查询，来了解一下这个过程。
3.1 数据写入
数据写入，以逻辑日志的形式，将原始 series 信息记录到 wal (json 格式)文件，随后周期到达后，会将此 json 批量转储成 parquet 文件，上传到 S3。包括记录 meta 信息、数据写入 json 文件。
将 series 的 schema 信息存储到 etcd
代码入口：
D:\opensource\openobserve\src\service\metrics\prom.rsremote_write set_schema_metadata(org_id, &metric_name, StreamType::Metrics, extra_metadata) .await .unwrap();
复制代码
可以在 etcd 中查询到 series 的 schema 信息，schema 只记录 series 的 lable-key，metrics name 等信息。
#etcd查看所有的keyetcdctl get / --prefix --keys-only /zinc/observe/schema/org_name/metrics/avalanche_metric_mmmmm_0_0...# etcd查看avalanche_metric_mmmmm_0_0的schema信息etcdctl get --prefix /zinc/observe/schema/org_name/metrics/avalanche_metric_mmmmm_0_0
[{"fields":[],"metadata":{"start_dt":"1695868345620467","prom_metadata":"{\"metric_type\":\"Gauge\",\"metric_family_name\":\"avalanche_metric_mmmmm_0_0\",\"help\":\"A tasty metric morsel\",\"unit\":\"\"}","created_at":"1695868345620467","end_dt":"1695868297371000"}},{"fields":[ {"name":"__hash__","data_type":"Utf8","nullable":true,"dict_id":0,"dict_is_ordered":false,"metadata":{}}, {"name":"__name__","data_type":"Utf8","nullable":true,"dict_id":0,"dict_is_ordered":false,"metadata":{}}, {"name":"_timestamp","data_type":"Int64","nullable":true,"dict_id":0,"dict_is_ordered":false,"metadata":{}},
复制代码
series 数据写入 wal
代码入口：
D:\opensource\openobserve\src\common\infra\wal.rs
pub fn write_file(let file = get_or_create(
impl RwFile {let file_name = format!("{thread_id}_{key}_{id}{}", FILE_EXT_JSON);
复制代码
通过上述流程，生成如下 wal 文件
json 文件内容，和下文的 parquet 文件，在内容上一致。
转成 parquet 格式并上传到 S3
周期到达后，会将 wal 文件(如果记录在内存，则从内存中提取逻辑日志信息）组装成 parquet 文件。关于 parquet 如何进行嵌套数据存储（更适合 trace 数据存储），可以参考相关 https://github.com/apache/parquet-format 和 https://blog.csdn.net/Night_ZW/article/details/108359619 更深入了解。代码入口：
D:\opensource\openobserve\src\job\files\memory.rsD:\opensource\openobserve\src\job\files\disk.rsmove_files_to_storage
复制代码
可以周期性的看到 ingester 有如下上传信息：
[2023-10-03T02:06:58Z INFO openobserve::job::files::disk] [JOB] convert disk file: data/wal/files/org_name/metrics/up/0_2023_09_28_00_keeping_7113098315302051840hFwd3b.json[2023-10-03T02:06:58Z INFO openobserve::job::files::disk] [JOB] File upload begin: disk: data/wal/files/org_name/metrics/up/0_2023_09_28_00_keeping_7113098315302051840hFwd3b.json[2023-10-03T02:06:58Z INFO openobserve::service::schema] service:schema:schema_evolution; org_id="org_name" stream_name="up" stream_type=Metrics min_ts=1695894790904000[2023-10-03T02:06:58Z INFO openobserve::job::files::disk] [JOB] disk file upload succeeded: files/org_name/metrics/up/2023/09/28/00/7113098315302051840hFwd3b.parquet
复制代码
上传到 S3 的 parquet 文件，解析出来如下：
3.2 数据查询
在 Web 触发一次 metric 查询，avalanche_metric_mmmmm_0_0{label_key_kkkkk_0="label_val_vvvvv_0"}
Querier 执行查询
Querier 侧的入口日志如下：
[2023-10-06T08:25:14Z INFO openobserve::service::promql::search::grpc::storage] promql:search:grpc:storage:create_context; org_id="org_name" stream_name="avalanche_metric_mmmmm_0_0"[2023-10-06T08:25:14Z INFO openobserve::service::promql::search::grpc::storage] promql:search:grpc:storage:get_file_list; org_id="org_name" stream_name="avalanche_metric_mmmmm0_0"[2023-10-06T08:25:14Z INFO openobserve::service::promql::search::grpc::wal] promql:search:grpc:wal:get_file_list; org_id="org_name" stream_name="avalanche_metric_mmmmm_0_0" time_range=(1696573200343000, 1696580700343000)[2023-10-06T08:25:14Z INFO actix_web::middleware::logger] 10.244.0.7 "GET /api/org_name/prometheus/api/v1/query_range?start=1696573500343000&end=1696580700343000&step=0&query=avalanche_metric_mmmmm_0_0%7Blabel_key_kkkkk_0%3D%22label_val_vvvvv_0%22%7D HTTP/1.1" 200 4356 "-" "http://127.0.0.1:5080/web/metrics?stream=avalanche_metric_mmmmm_0_12&period=2h&refresh=0&query=YXZhbGFuY2hlX21ldHJpY19tbW1tbV8wXzEye30=&org_identifier=org_name" "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/116.0" 0.063740
复制代码
代码逻辑流程如下：
create_context get_file_list // 获取包含metric的files列表 cache_parquet_files // 下载parquet文件 db::schema::get(org_id, stream_name, stream_type) // 从etcd获取当前metric最新的schema信息 register_table // 注册 datafusion 以适配parquet和json的数据查询
复制代码
首先从 filelist 中找到 metric 做为 stream name 的 files 列表，
files内容{"key":"files/org_name/metrics/avalanche_metric_mmmmm_0_1/2023/09/28/00/71129873952131031049nQPk2.parquet","meta":{"min_ts":1695868297371000,"max_ts":1695868942371000,"records":430,"original_size":274733,"compressed_size":9729},"deleted":false}
复制代码
然后去 S3 下载这些 parquet 文件。由于 metric 的 schema 信息可能会发生变化，但频率不高，所以从 etcd 获取到 metric 的所有 metric label 标签后，做为查询索引，去搜索 parquet 文件和当前 wal 文件中的内容。OpenObserve 使用 datafusion 将 json 和 parquet 数据一起进行 SQL 查询，进行 metric 的过滤。
D:\opensource\openobserve\src\service\search\datafusion\exec.rs
datasource::{ file_format::{ file_type::{FileType, GetExt}, json::JsonFormat, parquet::ParquetFormat,
复制代码
可以参考 datafusion 官方文档 https://arrow.apache.org/datafusion/ ，了解查询引擎如何将各类数据，进行 SQL 分析。
其他查询
OpenObserve 也提供了 series、labels、values 等查询接口，可以继续了解。
D:\opensource\openobserve\src\handler\http\request\prom\mod.rs#[get("/{org_id}/prometheus/api/v1/series")]pub async fn series_get( org_id: web::Path<String>, req: web::Query<meta::prom::RequestSeries>,) -> Result<HttpResponse, Error> { series(&org_id, req.into_inner()).await}
D:\opensource\openobserve\src\service\metrics\prom.rs
pub(crate) async fn get_series(
复制代码
评论