自定义 Attribute Store 运维

自定义 Attribute Store 会把选定的 Flow Attribute 写入 PostgreSQL。Dex 以异步方式逐个 Flow 写入最新状态，并采用至少一次尝试语义，因此每种目标写入都具有幂等性。

目前只支持 PostgreSQL 作为 Attribute Store。Redis 支持由 issue 616 跟踪。

Dex 不会创建、迁移或回填目标。请在启动 server 前创建 SQL 表。

配置

attributeStore:
  stores:
    reporting:
      type: postgres
      dsn: postgres://user:password@host/database
      tableName: public.flow_attributes
    orders:
      type: postgres
      dsn: postgres://user:password@host/database
      tableName: public.orders
      rowKey: order-{customer_id}-{order_id}
      columnNaming: snake_case
      attributeMaps:
        LineItems:
          tableName: public.order_line_items
          instanceKeyColumn: line_item_id
  schemaSyncInterval: 1m
  syncBatchSize: 100
  syncAttemptTimeout: 30s
  syncRetryPolicy:
    initialInterval: 100ms
    maximumInterval: 30s
    backoffCoefficient: 2
    maximumAttempts: 0
    totalDuration: 1h

每个 store 都必须配置 type: postgres、tableName 和 DSN（dsn 或环境变量）。 server-zd 绝不会记录 DSN。tableName 接受 table 或 schema.table。Dex 会解析并引用标识符，Attribute value 始终通过 driver 参数传递。每个键及其默认值见 Attribute Store 同步。

启动契约

server 在启动时连接、ping 并验证每个目标。任一目标失败都会导致启动失败，并关闭已经打开的 client。

未配置 rowKey 时，每张表必须只有一个可存储字符串的 primary key。Dex 会发现该 key，并把 Flow ID 写入其中。配置了 rowKey 时，primary key 必须正好是模板命名的那些列，并且每一列都必须是字符串列。

其他必填 SQL 列都必须有 default 或 generated value，这样 partial insert 才能创建新行。Attribute name 精确匹配其余的列；配置 columnNaming 时按其 snake_case 形式匹配。

Row key 与列名规则

store 接受两个可选配置。不配置时，store 把每个 Flow ID 写入唯一的 primary key 列。

rowKey 把每个 Flow ID 拆成 primary key 的各列。业务 ID 是 Flow ID 的一部分、需要按它查询时，或者多个 Flow type 描述同一个实体时使用它。模板由固定文字和占位符组成：

{column} 把捕获的文字写入这一 primary key 列。
{*} 只匹配文字，不写入任何列。
每个占位符捕获到下一段固定文字为止，最后一个占位符捕获 Flow ID 的剩余部分。
两个占位符之间必须有固定文字，任何捕获都不能为空。

例如，order-{customer_id}-{order_id} 把 order-c1-o-7 拆成 customer_id c1 和 order_id o-7。Flow ID 与模板不匹配时，Dex 记录错误日志并跳过整个 batch，因为重试无法改变结果。写入 key 列的 Attribute 会像写入 Flow ID 列一样被过滤。

多个 store 可以指向同一张表。每次 upsert 只设置本 batch 的列，所以两个 Flow type 可以写同一行的不同列。例如，customer-{customer_id} 和 onboarding-{customer_id}-{*} 两个 store 都写同一个客户的那一行。

columnNaming: snake_case 把每个 Attribute 写入其名称的 snake_case 形式对应的列。 Dex 在两种大写字母前插入下划线：跟在小写字母或数字后面的，以及位于缩写末尾、后面跟着小写字母的；然后把所有字母转为小写。DeploymentURL 变为 deployment_url， URLPath 变为 url_path，S3Bucket 变为 s3_bucket。同一 batch 中两个 Attribute name 映射到同一列时，Dex 会把两者都过滤掉。默认情况下，每个 Attribute 写入与其名称完全相同的列。

AttributeMap 表

attributeMaps 把 AttributeMap 的每个实例写成它自己那张表里的一行，这样就能用 SQL 查询 map 实例，例如一个订单的明细行或一个群组的成员。每个 key 是一个 AttributeMap 的名称：

tableName 选择实例表，写成 table 或 schema.table。
instanceKeyColumn 存放去掉转义后的实例 key：写成 LineItems/item%2F7 的实例存为 item/7。

实例表的 primary key 必须正好是 store 的 primary key 各列加上 instanceKeyColumn，并且都是字符串列。使用 order-{customer_id}-{order_id} 时，LineItems 表的 primary key 是 customer_id、order_id、 line_item_id。其他必填列都必须有 default 或 generated value。Dex 在启动时和每次 schema refresh 时检查实例表，和 store 自己的表一样。

写一个实例会替换它的行，删除实例会删除这一行。一个实例是一个完整的值，所以它的行始终等于最新的值：

JSON object 的每个顶层字段按 columnNaming 写入同名列。只有列能存整数时， JSON number 才作为整数写入。嵌套的 object 和 array 写入 JSON 或 JSONB 列。
其他任何值写入名为 value 的列。
值没有设置的其他非 key、非 generated 列，恢复为默认值；没有默认值时为 NULL。
没有对应列的字段会被跳过。实例表可以只存放每个值的一部分，所以 Dex 对每个这样的字段，每个进程只记录一次 warning。
映射到 primary key 列的字段会被跳过。只有它的值与 key 不同时，Dex 才记录 error。
无法转换的值记录为 error 并跳过。这一行仍然记录这个实例。

一个 batch 在同一个事务里写入 store 的行和它的全部实例行，每个实例取最新值。Dex 不在表之间创建外键：违反约束会让整个 batch 失败，并让这个 Flow 之后的写入一直等到重试预算用完。Flow 结束时，Dex 不会删除任何行。

选择了多个 store 的 Flow，只把每个 AttributeMap 写入声明了它的 store。其他 store 把它的实例当作普通 Attribute。

SQL 写入与 schema refresh

来自单个 Flow 的每个 batch 是一次 upsert。配置了 rowKey 时，upsert 写入每个 key 列，并按整个 primary key 处理冲突。Dex 不会合并不同 Flow 的 batch，也不维护 change log。

每个 store 独立 refresh。默认一分钟间隔的每轮会加入均匀的 90–110% jitter。 metadata 查询成功后会原子替换内存中的 immutable schema snapshot；查询失败时继续使用旧 snapshot，并在下个间隔后重试。

新增 nullable 或有 default 的列无需重启 server 即可被发现。一个 batch 要写的列不在当前 snapshot 里时（store 的表或 AttributeMap 表），它会先刷新这个 store 的 schema 再写。还没有被任何一次刷新确认缺失的列，立即刷新；已被确认缺失的列，等周期刷新再检查。 AttributeMap 表可以只存每个值的一部分，所以每个没有对应列的字段在一个进程里最多触发一次刷新。同时到达的 batch 会等这次刷新完成。因此，只要迁移在写这一列的 Worker 启动之前完成，新增列就不需要重启 server。删除或修改列后，后续 mutation 可能在本地验证时被过滤，也可能在 SQL 执行失败后重试。

SQL 类型契约

Attribute value	PostgreSQL 列
string	Character、varchar 或 text；检查声明长度
int64	有符号整数或 scale 为零的 numeric；检查范围和精度
double	浮点或 numeric
bool	Boolean
SDK datetime	Timestamp with time zone
JSON object	验证后的 JSON 或 JSONB
其他 object	bytea
null	仅 nullable 列

SDK datetime 可以使用 RFC3339 string 或 JSON epoch seconds。Dex 会拒绝过长的 string 或 binary、越界 integer、超过 decimal precision 的值、非有限浮点值和非法 JSON。

缺失列、类型不匹配、覆盖 Flow ID 或 key 列的写入、映射到同一列的多个 Attribute name，以及对必填列写入 null，都会逐字段记录并过滤。同一 batch 中的有效字段仍会写入。如果所有字段都被过滤，sync batch 成功结束且不执行 SQL。

交付与重试

Attribute Sync 保存最新状态，而不是 event history。发生不确定失败后，后续写入可能重复。因此 upsert 使用 Flow ID 作为幂等 key。

写入从不阻塞 Step、RPC 或 SetAttributes 的提交。失败的写入会重试到 syncRetryPolicy.totalDuration，之后丢弃并输出 error log。Dex 不会自动重放旧 Attribute，也不会为新配置的 store 自动 backfill。过滤写入、目标缺失和重试耗尽都会输出 error log。

fencing 表、关闭行为、重试和 metrics 详见 server-zd 运维指南中的 Attribute Store 同步。

配置​

启动契约​

Row key 与列名规则​

AttributeMap 表​

SQL 写入与 schema refresh​

SQL 类型契约​

交付与重试​

配置