兼容原理

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')
        return tokenStr

    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(t.authorization())

兼容 ES API 接口

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

兼容 ES DSL 语法

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

兼容 ES API 与标准 ES API 差异

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