KVClient编程示例

基本概念

openYuanrong datasystem (下文中称为数据系统)提供了近计算 KV 缓存能力，基于共享内存实现免拷贝的 KV 数据读写，实现高性能数据缓存。同时 KV 接口通过对接外部组件提供数据可靠性语义。

样例代码

Python

from yr.datasystem.ds_client import DsClient

client = DsClient("127.0.0.1", 31501)
client.init()

key = "key"
expected_val = b"value"
client.kv().set(key, expected_val)

val = client.kv().get([key])
assert val[0] == expected_val

client.kv().delete([key])

C++

#include <cstdio>
#include "datasystem/datasystem.h"

using namespace datasystem;

int main()
{
    ConnectOptions connectOptions = { .host = "127.0.0.1", .port = 31501 };
    auto client = std::make_shared<DsClient>(connectOptions);
    if (!client->Init().IsOk()) {
        fprintf(stderr, "Init failed\n");
        return -1;
    }

    std::string key = "testKey";
    std::string value = "Hello kv client";
    std::string value2 = "Hello modify";
    Status status = client->KV()->Set(key, value);
    if (!status.IsOk()) {
        fprintf(stderr, "Set failed\n");
        return -1;
    }

    std::string getValue;
    status = client->KV()->Get(key, getValue);
    if (!status.IsOk() || getValue != value) {
        fprintf(stderr, "Get failed\n");
        return -1;
    }

    status = client->KV()->Set(key, value2);
    if (!status.IsOk()) {
        fprintf(stderr, "Set failed\n");
        return -1;
    }

    status = client->KV()->Get(key, getValue);
    if (!status.IsOk() || getValue != value2) {
        fprintf(stderr, "Get failed\n");
        return -1;
    }

    status = client->KV()->Del(key);
    if (!status.IsOk()) {
        fprintf(stderr, "Del failed\n");
        return -1;
    }

    status = client->KV()->Get(key, getValue);
    if (!status.IsError()) {
        fprintf(stderr, "Get after delete should fail\n");
        return -1;
    }
    return 0;
}

使用限制

• key 仅支持大写字母、小写字母、数字以及如下特定字符：-_!@#%^*()+=:;。

• key 的最大长度为 1024 字节。

• value 的最大长度没有限制，但是不能超出配置的共享内存大小。

• 未写入二级缓存的数据，不保证数据可靠性，当发生故障时数据可能会丢失。

关于 KV 更多信息

数据一致性

KV 接口支持 Causal 级别数据读写一致性。 一致性模型定义参见 Consistency Models。

服务发现

客户端可以通过 ConnectOptions.serviceDiscovery 注入 IServiceDiscovery 实现，并通过 ServiceAffinityPolicy 指定 Worker 选择策略：

• PREFERRED_SAME_NODE（默认）：优先选择同节点 Worker；若无同节点 Worker，则随机选择。

• REQUIRED_SAME_NODE：必须选择同节点 Worker；若未匹配到同节点 Worker，则返回错误。

• RANDOM：随机选择 Worker。

使用节点亲和性时，需要让 Worker 和客户端都能读取同一套 host_id 信息：在 Worker 侧配置 host_id_env_name，并在客户端服务发现配置中设置相同的 hostIdEnvName。

通过 ETCD 发现 Worker

使用 ServiceDiscovery 从 ETCD 获取可用 Worker。如果 ETCD 启用了用户名和密码鉴权，还需要配置 username 和 password。

#include <cstdio>
#include "datasystem/datasystem.h"

using namespace datasystem;

int main()
{
    ServiceDiscoveryOptions sdOpts;
    sdOpts.etcdAddress = "127.0.0.1:2379";
    sdOpts.hostIdEnvName = "HOST_ID";
    sdOpts.affinityPolicy = ServiceAffinityPolicy::PREFERRED_SAME_NODE;
    sdOpts.username = "etcd_user";
    sdOpts.password = "etcd_password";

    auto serviceDiscovery = std::make_shared<ServiceDiscovery>(sdOpts);
    if (!serviceDiscovery->Init().IsOk()) {
        fprintf(stderr, "ServiceDiscovery Init failed\n");
        return -1;
    }

    ConnectOptions connectOptions;
    connectOptions.serviceDiscovery = serviceDiscovery;
    auto client = std::make_shared<DsClient>(connectOptions);
    if (!client->Init().IsOk()) {
        fprintf(stderr, "Init failed\n");
        return -1;
    }
    return 0;
}

通过 Coordinator 发现 Worker

使用 CoordinatorServiceDiscovery 从 Coordinator 获取可用 Worker。初始化时选择并固定访问一个 Coordinator 地址。

#include "datasystem/datasystem.h"

CoordinatorServiceDiscoveryOptions sdOpts;
sdOpts.serviceAddress = "127.0.0.1:31511";
sdOpts.hostIdEnvName = "HOST_ID";
sdOpts.affinityPolicy = ServiceAffinityPolicy::PREFERRED_SAME_NODE;

auto serviceDiscovery = std::make_shared<CoordinatorServiceDiscovery>(sdOpts);
ASSERT_TRUE(serviceDiscovery->Init().IsOk());

ConnectOptions connectOptions;
connectOptions.serviceDiscovery = serviceDiscovery;
auto client = std::make_shared<DsClient>(connectOptions);
ASSERT_TRUE(client->Init().IsOk());

如果 Coordinator 地址由外部服务发现系统提供，可以实现 ICoordinatorDiscovery 并注入 CoordinatorServiceDiscoveryOptions。Discovery 操作成功但暂无候选时可以返回空列表；当前 CoordinatorServiceDiscovery 会拒绝空列表，非空列表只缓存并使用首个地址。

class CustomCoordinatorDiscovery : public ICoordinatorDiscovery {
public:
    Status GetCoordinators(std::vector<std::string> &serviceList) override
    {
        serviceList = { "127.0.0.1:31511" };
        return Status::OK();
    }
};

CoordinatorServiceDiscoveryOptions sdOpts;
sdOpts.coordinatorDiscovery = std::make_shared<CustomCoordinatorDiscovery>();
auto serviceDiscovery = std::make_shared<CoordinatorServiceDiscovery>(sdOpts);
ASSERT_TRUE(serviceDiscovery->Init().IsOk());

Python KVClient 使用 service_discovery 参数接入服务发现，不需要预先指定 Worker 的 host 和 port。 当同时传入 service_discovery 与 host/port 时，KVClient 会使用 service_discovery 并忽略 host/port。 若期望同一个 KVClient 在当前连接的 Worker 故障后切换到其他已发现 Worker，需要同时设置 enable_cross_node_connection=True。

from yr.datasystem import (
    CoordinatorServiceDiscovery,
    CoordinatorServiceDiscoveryOptions,
    KVClient,
    ServiceAffinityPolicy,
)

options = CoordinatorServiceDiscoveryOptions()
options.service_address = "127.0.0.1:31511"
options.cluster_name = "cluster-a"
options.host_id_env_name = "HOST_ID"
options.affinity_policy = ServiceAffinityPolicy.PREFERRED_SAME_NODE

service_discovery = CoordinatorServiceDiscovery(options)
service_discovery.init()

client = KVClient(service_discovery=service_discovery, enable_cross_node_connection=True)
client.init()

数据溢出到磁盘

KV 数据存储在数据系统的共享内存中，当内存不足时，支持自动将数据溢出到磁盘并从内存中删除数据。当数据需要读取时，自动从磁盘中加载到共享内存。 若磁盘空间也不足时，如果数据已写入到二级缓存，则自动将数据从本地磁盘和内存中删除。当数据需要读取时，自动从二级缓存加载到共享内存。 使用 KV 溢出功能，需要在部署时指定相关参数，默认为关闭。

# The path of the spilling, empty means local_disk spill disabled.
# It will create a new subdirectory("datasystem_spill_data") under the SPILL_DIRECTORY to store the spill file.
# Example: If SPILL_DIRECTORY is "/home/spill", spill files will exist in the "/home/spill/datasystem_spill_data".
spillDirectory: ""

数据溢出有以下参数，可用于设置磁盘空间上限、溢出的并发线程、文件大小、是否可以迁移数据到远端内存等参数，用于性能调优。

# Maximum amount of spilled data that can be stored in the spill directory. If spill is enable and spillSizeLimit is 0, spillSizeLimit will be set to 95% of the spill directory.
# Unit for spillSizeLimit is Bytes.
spillSizeLimit: "0"
# It represents the maximum parallelism of writing files, more threads will consume more CPU and I/O resources.
spillThreadNum: 8
# The size limit of single spill file, spilling objects which larger than that value with one object per file.
# If there are some big objects, you can increase this value to avoid run out of inodes quickly.
# The valid range is 200-10240.
spillFileMaxSizeMb: 200
# The maximum number of open file descriptors about spill. If opened file exceed this value,
# some files will be temporarily closed to prevent exceeding the maximum system limit. You need reduce this value if your system resources are limited.
# The valid range is greater than or equal to 8.
spillFileOpenLimit: 512
# Disable readahead can mitigate the read amplification problem for offset read, default is true
spillEnableReadahead: true

内存均衡

memoryRebalance 用于开启由 master 调度的内存均衡能力。开启后，master 会根据各 worker 上报的共享内存使用情况，选择内存使用率较高的 worker 作为 source，选择内存使用率较低且仍有可用共享内存的 worker 作为 target，并下发对象迁移任务，将部分 KV/Object 缓存对象迁移到 target worker，以降低单个 worker 的共享内存压力。

该能力默认关闭。Kubernetes Helm 部署时可通过如下配置开启：

global:
  memoryRebalance:
    enabled: true
    # source worker 的共享内存使用率阈值（百分比），取值范围：1-100
    rebalanceSourceUsagePercent: 70
    # source 与 target 之间的最小使用率差值（百分比），取值范围：1-100
    rebalanceUsageGapPercent: 20
    # 失败或超时后的冷却时间（秒）
    rebalanceCooldownS: 60
    # 任务上报宽限时间（毫秒）
    rebalanceTaskReportGraceMs: 60000
    # 单次任务最多迁移数据量（字节），默认 1GB；超过 5 位需用字符串
    rebalanceMaxMigrateBytesPerRound: "1073741824"

当 enabled 为 false 时，master 不会调度内存均衡任务，worker 在本地共享内存不足时仍按既有驱逐、Spill 和二级缓存策略处理。开启该能力后，如果集群中没有满足条件的低使用率 worker，数据仍会继续走本地驱逐、Spill 或二级缓存流程，因此该配置不能替代 Spill 或二级缓存可靠性配置。

数据可靠性

数据系统 KV 接口提供可靠性语义，在数据写入时，通过 writeMode 参数配置数据可靠性级别。仅当 writeMode 配置为 WRITE_THROUGH_L2_CACHE 或 WRITE_BACK_L2_CACHE 时才保证数据可靠性，否则当出现故障或空间不足时，数据可能丢失。
数据系统通过对接外部存储组件作为二级缓存实现数据可靠性。当前支持的二级缓存组件有： distributed_disk / OBS / SFS。 在集群部署时，需要在数据系统的部署参数中配置二级缓存相关参数，若未配置，则在 KV 写入时 writeMode 参数指定为 WRITE_THROUGH_L2_CACHE 或 WRITE_BACK_L2_CACHE 时会写入失败。

集群部署时通过以下参数指定二级缓存类型。

# 指定二级缓存的类型。可选值为：'distributed_disk', 'obs', 'sfs'.
# 默认值为'none'，表示不支持二级缓存。
l2CacheType: "none"

对接各类外部组件的配置参数如下：

distributed_disk

distributedDisk:
  # The root path used by distributed disk storage.
  path: ""
  # The maximum size in MB of a single distributed disk data file.
  maxDataFileSizeMb: 1024
  # The maximum time interval in milliseconds before distributed disk group commit flushes pending writes.
  syncIntervalMs: 1000
  # The batch size threshold in bytes that triggers distributed disk group commit flush.
  syncBatchBytes: 33554432
  # The fixed interval in seconds between distributed disk background compact runs.
  compactIntervalS: 3600

obs

obs:
  # The access key for obs AK/SK authentication. If the value of encryptKit is not plaintext, encryption is required.
  obsAccessKey: ""
  # The secret key for obs AK/SK authentication. If the value of encryptKit is not plaintext, encryption is required.
  obsSecretKey: ""
  # OBS endpoint. Example: "xxx.hwcloudtest.cn"
  obsEndpoint: ""
  # OBS bucket name.
  obsBucket: ""
  # Whether to enable the https in obs. false: use HTTP (default), true: use HTTPS
  obsHttpsEnabled: false
  # Use cloud service token rotation to connect obs.
  cloudServiceTokenRotation:
    # Whether to use ccms credential rotation mode to access OBS, default is false. If is enabled, need to specify
    # iamHostName, identityProvider, projectId, regionId at least.
    # In addition, obsEndpoint and obsBucket need to be specified.
    enable: false
    # Domain name of the IAM token to be obtained. Example:  iam.example.com.
    iamHostName: ""
    # Provider that provides permissions for the ds-worker. Example: csms-datasystem.
    identityProvider: ""
    # Project id of the OBS to be accessed. Example: fb6a00ff7ae54a5fbb8ff855d0841d00.
    projectId: ""
    # Region id of the OBS to be accessed. Example: cn-beijing-4.
    regionId: ""
    # Whether to access OBS of other accounts by agency, default is false. If is true, need to specify tokenAgencyName
    # and tokenAgencyDomain.
    enableTokenByAgency: false
    # Agency name for proxy access to other accounts. Example: obs_access.
    tokenAgencyName: ""
    # Agency domain for proxy access to other accounts. Example: op_svc_cff.
    tokenAgencyDomain: ""

sfs

sfsTurbo:
  # Endpoint of sfs-turbo, which is used to concatenate the shared path in sfs-turbo. such as '172.21.7.239'.
  endpoint: ""
  # Sfs-turbo sub-path mounted to ds-worker. If this parameter is not specified, the root directory '/' of sfs-turbo
  # is mounted by default.
  subPath: ""
  # Specifies the sfs-turbo ID, which can be viewed on the sfs-turbo page.
  id: "0"
  # Specifies the sfs-turbo enterprise project ID, which can be viewed on the sfs-turbo page.
  projectId: "0"
  # Specifies the capacity of using sfs-turbo. Note that the size must be smaller than the size of sfsTurbo.
  capacity: "500Gi"