兼容原理

Elasticsearch 是一款基于 Apache Lucene 的开源、分布式搜索和分析引擎，支持实时存储结构化、非结构化和向量数据，提供快速的混合和向量搜索。

日志服务提供兼容 Elasticsearch API，通过将标准 Elasticsearch DSL 查询翻译为日志服务的索引查询和 SQL 分析查询，并按照标准 Elasticsearch API 格式规范返回查询分析结果，从而实现与标准 Elasticsearch 查询协议的兼容。

兼容 ES API 访问域名

日志服务的兼容 Elasticsearch API 访问域名如下：

格式：https://${tls-endpoint}/es/

示例：https://tls-cn-beijing.volces.com/es/

兼容 ES API 访问使用的账号密码

日志服务的兼容 Elasticsearch API 访问的账号信息示例如下：

import hashlib

import hmac

import base64

class Util:

@staticmethod

def hmac(key, string):

return hmac.new(key, string.encode('utf-8'), hashlib.sha256).digest()

@staticmethod

def sha256(data):

return hashlib.sha256(data.encode('utf-8')).digest()

class tlsSigners:

def __init__(self,ak,sk):

self.service_name = "TLS"

self.region = "cn-north-1"

self.algorithm = "HMAC-SHA256"

self.v4Identifier = "request"

self.ak = ak

self.sk = sk

self.body = "{}"

self.path = "/es/"

self.method = "POST"

self.query = ""

self.datetime = "20250101T000000Z"

self.request = {"headers": ["x-content-sha256","x-date"]}

def authorization(self):

cred_string = self.create_scope()

parts = [

f"HMAC-SHA256 Credential={self.ak}/{cred_string}",

f"SignedHeaders={self.signed_headers()}",

f"Signature={self.signature()}"

]

token = ', '.join(parts)

tokenBytes = "{}".format(token).encode('utf-8')

tokenStr = base64.b64encode(tokenBytes).decode('utf-8')

authorizationStr = base64.b64encode((self.ak + ":" + tokenStr).encode('utf-8')).decode('utf-8')

return authorizationStr

def signature(self):

signing_key = self.get_signing_key()

sign_str = self.get_sign_str()

return Util.hmac(signing_key,sign_str).hex()

def get_signing_key(self):

k_date = Util.hmac(f"{self.sk}".encode('utf-8'),self.datetime[:8])

region = self.region

k_region = Util.hmac(k_date,region)

k_service = Util.hmac(k_region,self.service_name)

sign_key = Util.hmac(k_service,self.v4Identifier)

return sign_key

def get_sign_str(self):

parts = [

self.algorithm,

self.datetime,

self.create_scope(),

Util.sha256(self.get_canonical_string()).hex()

]

return '\n'.join(parts)

def get_canonical_string(self):

parts = [

self.method,

self.path,

self.query,

"x-content-sha256:"+Util.sha256(self.body).hex(),

"x-date:" + self.datetime,

"",

self.request["headers"][0]+";"+self.request["headers"][1],

Util.sha256(self.body).hex(),

]

return '\n'.join(parts)

def signed_headers(self):

headers = sorted(k.lower() for k in self.request['headers'])

return ';'.join(headers)

def create_scope(self):

return '/'.join( [self.datetime[0:8], self.region, self.service_name, self.v4Identifier])

if __name__ == '__main__':

t = tlsSigners("your_ak","your_sk")

print("Authorization: Basic "+t.authorization())

兼容 ES API 接口

日志服务的兼容 Elasticsearch API 支持兼容如下标准 Elasticsearch API。

请求方法	ES API 路径	说明
POST	/${index}/_search	查询指定的 Index。 Index 对应日志服务的 Project 和 Topic，格式：${project}.${topic_id}。
POST	/${index}/_async_search	批量查询。
POST	/_msearch	批量查询。
POST	/${index}/_bulk	写入数据到指定的 Index。不支持更新数据，仅支持写入数据。
GET	/${index}/_mapping	获取指定 Index 的 Mapping。

兼容 ES DSL 语法

日志服务的兼容 Elasticsearch API 支持如下标准 Elasticsearch DSL。

DSL 分类	日志服务兼容的 DSL 语法
查询类	bool：布尔运算 regexp：正则表达式匹配（翻译为日志服务的 SQL 正则函数，返回配置了索引的字段） match_all：匹配所有日志 match：索引匹配 match_phrase：短语匹配（翻译为日志服务的 SQL LIKE 语法，返回配置了索引的字段） multi_match：多字段匹配 prefix：前缀查询（翻译为日志服务的 SQL LIKE 语法，返回配置了索引的字段） range：范围查询 term：关键字查询（翻译为日志服务的索引查询语法） terms：多个关键字查询（翻译为日志服务的索引查询语法） wildcard：通配符匹配（翻译为日志服务的 SQL LIKE 语法，返回配置了索引的字段） exists：字段存在查询（翻译为 key:* 查询）
聚合类	分组统计： Date histogram：根据@timestamp字段进行直方图统计 Histogram：根据数值间隔进行直方图统计 Range：根据数值范围进行分组统计 Terms：根据字段唯一值进行分组统计 Filter：根据单个过滤条件进行分组统计 Filters：根据多个过滤条件进行分组统计指标统计： Min：统计最小值 Max：统计最大值 Avg：统计平均值 Sum：统计总和 Count：统计总数返回日志： Top hits：返回分组后的日志

兼容 ES API 与标准 ES API 差异

日志服务的兼容 Elasticsearch API 与标准 Elasticsearch API差异如下：

差异点	标准 Elasticsearch API	日志服务兼容 Elasticsearch API
更新数据	支持	不支持
包含 Script 的查询语句	支持	不支持
自动创建索引	支持	不支持需通过 API 或在日志服务控制台手动创建。
PB 级数据量 Topic	不支持	支持不需要做 Index 滚动。
字段映射	@timestamp	__time__
字段映射		_id	日志服务数据中已存在_id字段：是：使用 _id。否：使用 uuid 或 _pack_meta。