复制链接或点击阅读原文，查看回放视频

https://www.bilibili.com/video/BV1AM411b7E3/?vd_source=8217e32e9012f691b56ca71735c1a472

1. 从可观测性建设角度出发，总结大家在日常工作中遇到的痛点；
2. 介绍 DeepFlow 的软件架构、系统组成；
3. 讲解 DeepFlow 的关键特性 AutoTagging 技术；
4. 讲解支撑 AutoTagging 10x 性能提升的 SmartEncoding 技术；
5. 总结并分享后续的迭代和演进计划。

01

观测数据存储的挑战

02

DeepFlow 软件架构

03

AutoTagging：构建标准化的标签体系

• 自定义标签
• k8s.label/k8s.env/k8s.annotation/..
• os.app/os.proc/...
• cloud.tag
• 进程
• 进程名
• 云资源
• 资源池：区域/可用区
• 计算资源：云服务器/宿主机
• 网络资源：VPC/子网/路由器/IP地址
• 网络服务：安全组/负载均衡器/NAT网关/对等连接/云企业网
• 存储资源：云数据库RDS/Redis
• 容器资源
• 容器集群/容器节点/命名空间/Ingress/容器服务/工作负载/POD

• 为什么是 Agent watch 并上报 K8s资源？一个 Server 可以管理多个集群中的 Agent，Agent在所属集群中watch K8s，避免了集群外部用 server watch 时涉及到的权限和配置问题。

    
  

• 如何控制 Agent 对 K8s 资源的 watch？避免 K8s 的 API 压力过大，不能让所有的 Agent 都去 watch K8s，Server 在每个集群中选举一个 Agent；仅让被选中的 Agent watch K8s 资源。

• 内存优化：Agent 仅抓取同步必须的字段，同时会第一时间进行压缩。
• 带宽优化：仅当 K8s 资源有变化时，Agent 才会向 Server 发送具体的资源信息

• 同步云资源信息：通过调用云平台 API 进行资源抽象和转换，然后将相关标签信息保存至 MySQL 中，并定期更新 ClickHosue 中的字典。

   

• 同步 Legacy Host 信息：一些环境中，可能没有真正意义的云平台，或者存在一些传统主机需要监控，这就需要用Legacy Host同步方案。由于没有具体的云API，我们完全通过Agent抓取所在服务器的名称等基本信息和网卡信息，上报给Server汇总并进行资源抽象。

   

• 同步托管 K8s 信息：当 K8s 平台部署在云资源上时，要做到真正的可观测性，需要将K8s的资源和云资源关联起来，才能真正做到无缝地关联、切分和下钻。我们一方面通过获取 K8s 资源所在的 VPC，基于 VPC 内 IP 的唯一性，通过 VPC + IP 将 K8s 的容器节点与云服务器关联起来；另一方面通过将云平台的 API 调用与 K8s 独立，两者使用不同的调用频率，从而解决大规模场景下，云平台 API 慢与 K8s 资源更新快的矛盾。

04

SmartEncoding：实现 10x 性能提升

采集时编码

存储时编码

查询时编/解码

SELECT col_1, col_2, col_3 \
FROM   tbl_1 \
WHERE  col_4 = y \
GROUP BY col_1, col_2 \
HAVING   col_5 > 100 \
ORDER BY col_3 \
LIMIT 100
    ```
我们可以查询某个 Tag 的所有候选项：
```sql
SHOW tag ${tag_name} values FROM ${table_name}

SHOW tag ${tag_name} values
FROM ${table_name}
WHERE display_name LIKE '*abc*'

SELECT pod
FROM `vtap_flow_port.1m`
WHERE pod_cluster = 'cluster1'
GROUP BY pod

更多详细用法^[4]

查询 Universal Tag

• ClickHouse 的观测数据表中保存tag ID

    

  CREATE TABLE flow_metrics.`vtap_flow_port.1m`
  (
      `time` DateTime('Asia/Shanghai') COMMENT 'v6.1.8' CODEC(DoubleDelta),
      `ip4` IPv4 COMMENT 'IPv4地址',
      `ip6` IPv6 COMMENT 'IPV6地址',
      `is_ipv4` UInt8 COMMENT '是否IPV4地址. 0: 否, ip6字段有效, 1: 是, ip4字段有效',
      `l3_device_id` UInt32 COMMENT 'ip对应的资源ID',
      `l3_device_type` UInt8 COMMENT 'ip对应的资源类型',
      `l3_epc_id` Int32 COMMENT 'ip对应的EPC ID',
      `pod_cluster_id` UInt16 COMMENT 'ip对应的容器集群ID',
      `pod_group_id` UInt32 COMMENT 'ip对应的容器工作负载ID',
      `pod_id` UInt32 COMMENT 'ip对应的容器POD ID',
      `pod_node_id` UInt32 COMMENT 'ip对应的容器节点ID',
      `pod_ns_id` UInt16 COMMENT 'ip对应的容器命名空间ID'
  )
  ENGINE = Distributed(...)
  
  

  
  

• ClickHouse 的字典表中保存tag ID和名称对应关系

    

  
  CREATE DICTIONARY flow_tag.pod_map
  (
      `id` UInt64,
      `name` String,
      `icon_id` Int64
  )
  PRIMARY KEY id
  SOURCE(...)
  
  

    

• 通过 dictGet 实现tag ID到名称的转换

     

  SELECT dictGet(flow_tag.pod_map, 'name', toUInt64(pod_id)) AS pod
  FROM `vtap_flow_port.1m`
  WHERE pod = 'deepflow'
  GROUP BY pod
  LIMIT 1
  
  

查询 K8s label

• ClickHouse 的字典表中保存tag ID和 K8s label对应关系

    

  CREATE DICTIONARY flow_tag.k8s_label_map
  (
      `pod_id` UInt64,
      `key` String,  
      `value` String,
  )
  PRIMARY KEY pod_id, key
  SOURCE(...)
  LIFETIME(MIN 0 MAX 60)
  LAYOUT(COMPLEX_KEY_HASHED())
  
  

    

• 通过 dictGet 实现tag ID到 K8s label 的转换

     

  SELECT dictGet(flow_tag.k8s_label_map, 'value', (toUInt64(pod_id), 'app')) AS `label.app`
  FROM `vtap_flow_port.1m`
  WHERE `label.app` = 'xxx'
  LIMIT 1
  
  

• 存储集成数据时，会将数据中原有的 Tag 和 Metric 的 name 和 value 分别定义为 Array 类型，一一对应。

     
  

  CREATE TABLE ext_metrics.prometheus_web
  (
      `time` DateTime('Asia/Shanghai') CODEC(DoubleDelta),
      `_tid` UInt8 COMMENT '用于区分trident不同的pipeline',
      `az_id` UInt16 COMMENT '可用区ID',
      `host_id` UInt16 COMMENT '宿主机ID',
      `tag_names` Array(String) COMMENT '额外的tag',
      `tag_values` Array(String) COMMENT '额外的tag对应的值',
      `metrics_float_names` Array(String) COMMENT '额外的metrics',
      `metrics_float_values` Array(Float64) COMMENT '额外的metrics值'
  )
  
  

     

• Tag 候选项只需要保留不重复的值，所以我们使用 ReplacingMergeTree Engine

     

  CREATE TABLE flow_tag.ext_metrics_custom_field_local
  (
      `time` DateTime('Asia/Shanghai') CODEC(DoubleDelta),
      `table` LowCardinality(String),
      `vpc_id` Int32,
      `pod_ns_id` UInt16,
      `field_type` LowCardinality(String) COMMENT 'value: tag, metrics',
      `field_name` LowCardinality(String),
      `field_value_type` LowCardinality(String) COMMENT 'value: string, float'
  )
  ENGINE = ReplacingMergeTree(time)
  
  

     

• 通过 indexOf 进行 name 和 value 的对应

     
  

  SELECT tag_values[indexOf(tag_names, 'host')] AS `tag.host`
  FROM deepflow_agent_collect_sender
  WHERE (tag_values[indexOf(tag_names, 'host')]) = 'xxxx'
  LIMIT 1
  
  

05

总结与后续迭代计划

06

关于DeepFlow

参考资料