源码分析 kubernetes apisix ingress crd 及缓存的实现原理 (二)
篇幅原因分成两个两篇, 第一篇主要分析了 apisix ingress controller 的架构实现以及对 k8s ingress 和 endpoints 进行分析.
源码分析 kubernetes apisix ingress controller 控制器的实现原理 (一)
本文也第二篇, 主要分析 apisix ingress 对自定义 CRD 资源的处理, 以及 apisix ingress 里如何建立多索引的缓存系统.
apisix 内置 crd provider 实现原理
前面分析了 apisix-ingress 对 k8s ingress 和 endpoints 资源处理的过程. 但 ingress 结构比较表现力有限, 通过 configmap 和 annotation 扩充的配置不够友好. 所有 apisix 也支持自定义的资源类型 ApisixRoute 和 ApisixUpstream. 这两个抽象过的 crd 的资源在配置上更加的灵活丰富, 也更好理解. 当前社区中除了 nginx-ingress 外, 其他有影响力的 ingress 都可通过自定义 CRD 来配置, 毕竟 ingerss 结构体略显简单.
不管使用 ingress 还是自定义 crd (ApisixRoute, ApisxiUpstream) , 控制器都会先把配置拆分组装成 apisix 自身的配置项. 比如会把 k8s ingress 对应到 apisix 的 routes/upstream/pulugin/ssl 四个配置, 而 apisixroute 会把配置拆分组合成 routes/stream_routes/upstream/plugin 四个配置. 在配置构建完成后, 依次向 apisix admin 发起变更请求.
实例化
ApisixRouteController 控制器内部实例化了 apisixRoute, service, apisixUpstream 这三个类型的 Informer. 在 apisixRouteInfomer 注册的 eventHandler 事件方法, 其逻辑就是根据增删改事件, 向 workqueue 发送不同的 event.
而在 service 和 apisixUpstream 注册的 eventHandler 事件方法, 其逻辑就是根据增删改事件, 向 relatedWorkqueue 发送不同的 event.
func newApisixRouteController(common *apisixCommon) *apisixRouteController {
c := &apisixRouteController{
// 只有一个协程
workers: 1,
// 索引关系, 通过 serviceName 找到 apisixroute key
svcMap: make(map[string]map[string]struct{}),
// 索引关系, 通过 upstream name 找到 apisixroute key
apisixUpstreamMap: make(map[string]map[string]struct{}),
}
// apisixRoute informer
c.ApisixRouteInformer.AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc: c.onAdd,
UpdateFunc: c.onUpdate,
DeleteFunc: c.onDelete,
},
)
// service informer
c.SvcInformer.AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc: c.onSvcAdd,
},
)
// apisix upstream informer
c.ApisixUpstreamInformer.AddEventHandler(
cache.ResourceEventHandlerFuncs{
AddFunc: c.onApisixUpstreamAdd,
UpdateFunc: c.onApisixUpstreamUpdate,
},
)
return c
}
启动
启动两组协程, 分别是 runWorker 和 runRelatedWorker.
func (c *apisixRouteController) run(ctx context.Context) {
log.Info("ApisixRoute controller started")
defer log.Info("ApisixRoute controller exited")
defer c.workqueue.ShutDown()
defer c.relatedWorkqueue.ShutDown()
for i := 0; i < c.workers; i++ {
// 处理 apisixRoute 对象
go c.runWorker(ctx)
// 处理 service 和 apisixUpstream 对象
go c.runRelatedWorker(ctx)
}
<-ctx.Done()
}
runWorker() 方法监听 workqueue 队列, 然后处理 apisixRoute 的配置同步. 而 runRelatedWorker 则监听 relatedWorkqueue 队列, 处理 servcie 和 apisixUpstream 两个资源类型, 维护 service -> apisixroutekey 和 upstreamkey -> apisixroutekey 的映射关系, 还会往 workqueue 传递通知, 毕竟 apisixRoute 资源是跟 service 和 apisixUpstream 关联的, 当 service 和 apisixUpstream 发生变更时, apisix 的配置也需要变更下.
type apisixRouteController struct {
// service key -> apisix route key
svcMap map[string]map[string]struct{}
// apisix upstream key -> apisix route key
apisixUpstreamMap map[string]map[string]struct{}
}
func (c *apisixRouteController) runWorker(ctx context.Context) {
for {
obj, quit := c.workqueue.Get()
if quit {
return
}
switch val := obj.(type) {
case *types.Event:
err := c.sync(ctx, val)
c.workqueue.Done(obj)
}
}
}
func (c *apisixRouteController) runRelatedWorker(ctx context.Context) {
for {
obj, quit := c.relatedWorkqueue.Get()
if quit {
return
}
ev := obj.(*routeEvent)
switch ev.Type {
case "service":
err := c.handleSvcAdd(ev.Key)
...
case "ApisixUpstream":
err := c.handleApisixUpstreamChange(ev.Key)
...
}
}
}
sync 同步逻辑
sync 核心的同步配置流程. 先通过 apisix route lister 缓存中获取对象, 然后把结构转换成内部使用的结构, 最后调用 SyncManifests 方法, 把配置数据刷新到 apisix admin 里.
func (c *apisixRouteController) sync(ctx context.Context, ev *types.Event) error {
obj := ev.Object.(kube.ApisixRouteEvent)
namespace, name, err := cache.SplitMetaNamespaceKey(obj.Key)
if err != nil {
return err
}
var (
ar kube.ApisixRoute
tctx *translation.TranslateContext
)
switch obj.GroupVersion {
case config.ApisixV2beta3:
// 从 apisixRoute v2beta3 lister 缓存里获取对象
ar, err = c.ApisixRouteLister.V2beta3(namespace, name)
case config.ApisixV2:
// 从 apisixRoute v2 lister 缓存里获取对象
ar, err = c.ApisixRouteLister.V2(namespace, name)
default:
// 未知版本
return fmt.Errorf("unknown ApisixRoute version %v", obj.GroupVersion)
}
if err != nil {
return err
}
// 维护索引关系, apisix upstream key -> apisix route key
c.syncRelationship(ev, obj.Key, ar)
// 如果事件类型为删除, 则标记删除
if ev.Type == types.EventDelete {
ar = ev.Tombstone.(kube.ApisixRoute)
}
// 把当前的 apisixRoute 结构转成内部使用的 translation.TranslateContexT 结构
switch obj.GroupVersion {
case config.ApisixV2beta3:
...
case config.ApisixV2:
if ev.Type != types.EventDelete {
if err = c.checkPluginNameIfNotEmptyV2(ctx, ar.V2()); err == nil {
tctx, err = c.translator.TranslateRouteV2(ar.V2())
}
} else {
tctx, err = c.translator.GenerateRouteV2DeleteMark(ar.V2())
}
default:
// 未知版本
return fmt.Errorf("unknown ApisixRoute version %v", obj.GroupVersion)
}
// tctx 从 apisixRoute 结构中解析出 apisix 的 Route/Upstream/StreamRoutes/Plugin 配置.
m := &utils.Manifest{
Routes: tctx.Routes,
Upstreams: tctx.Upstreams,
StreamRoutes: tctx.StreamRoutes,
PluginConfigs: tctx.PluginConfigs,
}
var (
added *utils.Manifest
updated *utils.Manifest
deleted *utils.Manifest
)
// 判断事件类型赋值到不同的对象上.
if ev.Type == types.EventDelete {
deleted = m
} else if ev.Type == types.EventAdd {
added = m
} else {
oldCtx, _ := c.translator.TranslateOldRoute(obj.OldObject)
om := &utils.Manifest{
Routes: oldCtx.Routes,
Upstreams: oldCtx.Upstreams,
StreamRoutes: oldCtx.StreamRoutes,
PluginConfigs: oldCtx.PluginConfigs,
}
added, updated, deleted = m.Diff(om)
}
// 使用 http 把配置更新到 apisix
return c.SyncManifests(ctx, added, updated, deleted)
}
TranslateRouteV2 结构体转换
TranslateRouteV2 用来解析 ApisixRoute 数据结构到 TranslateContext 里.
func (t *translator) TranslateRouteV2(ar *configv2.ApisixRoute) (*translation.TranslateContext, error) {
ctx := translation.DefaultEmptyTranslateContext()
// 组装 http 数据结构
if err := t.translateHTTPRouteV2(ctx, ar); err != nil {
return nil, err
}
// 组装 tcp/udp 数据结构
if err := t.translateStreamRouteV2(ctx, ar); err != nil {
return nil, err
}
return ctx, nil
}
下面是 translateHTTPRouteV2 拼装结构的过程.
func (t *translator) translateHTTPRouteV2(ctx *translation.TranslateContext, ar *configv2.ApisixRoute) error {
ruleNameMap := make(map[string]struct{})
for _, part := range ar.Spec.HTTP {
if _, ok := ruleNameMap[part.Name]; ok {
return errors.New("duplicated route rule name")
}
ruleNameMap[part.Name] = struct{}{}
// 设置超时时间
var timeout *apisixv1.UpstreamTimeout
if part.Timeout != nil {
timeout = &apisixv1.UpstreamTimeout{
Connect: apisixv1.DefaultUpstreamTimeout,
Read: apisixv1.DefaultUpstreamTimeout,
Send: apisixv1.DefaultUpstreamTimeout,
}
...
}
// 设置插件配置
pluginMap := make(apisixv1.Plugins)
// add route plugins
for _, plugin := range part.Plugins {
if !plugin.Enable {
continue
}
if plugin.Config != nil {
if plugin.SecretRef != "" {
sec, err := t.SecretLister.Secrets(ar.Namespace).Get(plugin.SecretRef)
if err != nil {
break
}
for key, value := range sec.Data {
plugin.Config[key] = string(value)
}
}
pluginMap[plugin.Name] = plugin.Config
} else {
pluginMap[plugin.Name] = make(map[string]interface{})
}
}
...
// 设置 remote addr
if err := translation.ValidateRemoteAddrs(part.Match.RemoteAddrs); err != nil {
return err
}
// 设置 route 对象的属性
route := apisixv1.NewDefaultRoute()
route.Name = apisixv1.ComposeRouteName(ar.Namespace, ar.Name, part.Name)
route.ID = id.GenID(route.Name)
route.Priority = part.Priority
route.RemoteAddrs = part.Match.RemoteAddrs
route.Vars = exprs
route.Hosts = part.Match.Hosts
route.Uris = part.Match.Paths
route.Methods = part.Match.Methods
route.EnableWebsocket = part.Websocket
route.Plugins = pluginMap
route.Timeout = timeout
if part.PluginConfigName != "" {
route.PluginConfigId = id.GenID(apisixv1.ComposePluginConfigName(ar.Namespace, part.PluginConfigName))
}
for k, v := range ar.ObjectMeta.Labels {
route.Metadata.Labels[k] = v
}
ctx.AddRoute(route)
backends := part.Backends
// 处理 backends 主机列表到 upstream 结构里.
if len(backends) > 0 {
backend := backends[0]
backends = backends[1:]
// 获取 service clusterip 和端口
svcClusterIP, svcPort, err := t.GetServiceClusterIPAndPort(&backend, ar.Namespace)
// 通过参数组件 upstream name
upstreamName := apisixv1.ComposeUpstreamName(ar.Namespace, backend.ServiceName, backend.Subset, svcPort, backend.ResolveGranularity)
// 通过 hash/crc32 生成 id
route.UpstreamId = id.GenID(upstreamName)
if len(backends) > 0 {
weight := translation.DefaultWeight // 默认为 100
// 配置自定义的 weight 权重值
if backend.Weight != nil {
weight = *backend.Weight
}
plugin, err := t.translateTrafficSplitPlugin(ctx, ar.Namespace, weight, backends)
if err != nil {
return err
}
// 配置自定义流量切分规则
route.Plugins["traffic-split"] = plugin
}
if !ctx.CheckUpstreamExist(upstreamName) {
// 生成 upstream 结构
ups, err := t.translateService(ar.Namespace, backend.ServiceName, backend.Subset, backend.ResolveGranularity, svcClusterIP, svcPort)
if err != nil {
return err
}
// 添加到 tctx 的 upstream 集合里
ctx.AddUpstream(ups)
}
}
// 下面忽略的代码是处理 externalName service 的场景, 忽略了.
...
}
return nil
}
至于 translateStreamRouteV2 的实现跟 translateHTTPRouteV2 大同小异, 不做分析.
SyncManifests
在第一篇中有详细的分析过 SyncManifests 实现原理, 这里不再详细复述.
简单说就是依次判断各个子配置是否为空, 对不为 nil 的配置进行处理, 处理的逻辑是通过 http restful 对 apifix admin 进行变更请求.
代码位置: pkg/providers/utils/manifest.go
func SyncManifests(ctx context.Context, apisix apisix.APISIX, clusterName string, added, updated, deleted *Manifest) error {
var merr *multierror.Error
if added != nil {
for _, u := range added.Upstreams {
if _, err := apisix.Cluster(clusterName).Upstream().Create(ctx, u); err != nil {
merr = multierror.Append(merr, err)
}
}
for _, pc := range added.PluginConfigs {
if _, err := apisix.Cluster(clusterName).PluginConfig().Create(ctx, pc); err != nil {
merr = multierror.Append(merr, err)
}
}
for _, r := range added.Routes {
if _, err := apisix.Cluster(clusterName).Route().Create(ctx, r); err != nil {
merr = multierror.Append(merr, err)
}
}
for _, sr := range added.StreamRoutes {
if _, err := apisix.Cluster(clusterName).StreamRoute().Create(ctx, sr); err != nil {
merr = multierror.Append(merr, err)
}
}
}
if updated != nil {
for _, r := range updated.Upstreams {
if _, err := apisix.Cluster(clusterName).Upstream().Update(ctx, r); err != nil {
merr = multierror.Append(merr, err)
}
}
for _, pc := range updated.PluginConfigs {
if _, err := apisix.Cluster(clusterName).PluginConfig().Update(ctx, pc); err != nil {
merr = multierror.Append(merr, err)
}
}
for _, r := range updated.Routes {
if _, err := apisix.Cluster(clusterName).Route().Update(ctx, r); err != nil {
merr = multierror.Append(merr, err)
}
}
for _, sr := range updated.StreamRoutes {
if _, err := apisix.Cluster(clusterName).StreamRoute().Update(ctx, sr); err != nil {
merr = multierror.Append(merr, err)
}
}
}
if deleted != nil {
...
}
if merr != nil {
return merr
}
return nil
}
apisix ingress 多级索引缓存设计
apisix ingress 内部使用 go-memdb 来构建多索引的缓存. go-memdb 是 hashicorp 社区的一个项目, 该库实现了实现内存级数据库, 支持丰富的索引, 支持多表的事务和多版本控制 mvcc 等.
go-memdb 的项目地址:
https://github.com/hashicorp/go-memdb
为什么使用 go-memdb 实现缓存系统 ?
因为 go-memdb 作为数据库是支持索引的, 且索引类别很是丰富, 不仅单字段索引, 类似 mysql 的联合索引, 另外如果索引字段为数字, 还可以 range 范围查询.
可以想象如果不使用 go-memdb, 而使用自定义索引映射, 那会相当的麻烦. 比如你的 struct 有 3 个字段, 后面想通过这三个字段的值直接找到对应的对象, 当然不能粗暴遍历, 通常需要多个 map[struct]interface{} 自定义索引关系, 插入还好, 更麻烦的是当触发更新和删除时, 需要维护已建立的索引.
一句话, 手动维护索引关系会相当麻烦.
通过下面的对象映射图, 应该让大家对多索引缓存的设计有更好的理解.
cache 实现原理
apisix ingress cache 的 schema 的代码位置.
在 db 里实例化了多个表结构, 每个表里又实例化多个索引映射. 当对 db 进行读写删操作时, go-memdb 会自动创建索引.
代码位置: pkg/apisix/cache/schema.go
_schema = &memdb.DBSchema{
Tables: map[string]*memdb.TableSchema{
"route": {
Name: "route",
Indexes: map[string]*memdb.IndexSchema{
"id": {
Name: "id",
Unique: true,
Indexer: &memdb.StringFieldIndex{Field: "ID"},
},
"name": {
Name: "name",
Unique: true,
Indexer: &memdb.StringFieldIndex{Field: "Name"},
AllowMissing: true,
},
"upstream_id": {
Name: "upstream_id",
Unique: false,
Indexer: &memdb.StringFieldIndex{Field: "UpstreamId"},
AllowMissing: true,
},
...
},
},
"upstream": {
Name: "upstream",
Indexes: map[string]*memdb.IndexSchema{
"id": {
Name: "id",
Unique: true,
Indexer: &memdb.StringFieldIndex{Field: "ID"},
},
"name": {
Name: "name",
Unique: true,
Indexer: &memdb.StringFieldIndex{Field: "Name"},
AllowMissing: true,
},
},
},
...
}
}
dbcache 里面实现了很多 apisix 内置对象的缓存管理, 下面拿 upstream 结构的缓存举例说明.
源码位置: pkg/apisix/cache/memdb.go
type dbCache struct {
db *memdb.MemDB
}
func NewMemDBCache() (Cache, error) {
db, err := memdb.NewMemDB(_schema)
if err != nil {
return nil, err
}
return &dbCache{
db: db,
}, nil
}
// 往 upstream 表里写对象
func (c *dbCache) InsertUpstream(u *v1.Upstream) error {
return c.insert("upstream", u.DeepCopy())
}
// 调用 memdb 的 insert 写数据
func (c *dbCache) insert(table string, obj interface{}) error {
txn := c.db.Txn(true)
defer txn.Abort()
if err := txn.Insert(table, obj); err != nil {
return err
}
txn.Commit()
return nil
}
// 从 upstream 表里获取数据
func (c *dbCache) GetUpstream(id string) (*v1.Upstream, error) {
obj, err := c.get("upstream", id)
if err != nil {
return nil, err
}
return obj.(*v1.Upstream).DeepCopy(), nil
}
// tnx.first 是获取第一条数据
func (c *dbCache) get(table, id string) (interface{}, error) {
txn := c.db.Txn(false)
defer txn.Abort()
obj, err := txn.First(table, "id", id)
if err != nil {
if err == memdb.ErrNotFound {
return nil, ErrNotFound
}
return nil, err
}
if obj == nil {
return nil, ErrNotFound
}
return obj, nil
}
// txn.get 可以获取多条数据
func (c *dbCache) list(table string) ([]interface{}, error) {
txn := c.db.Txn(false)
defer txn.Abort()
iter, err := txn.Get(table, "id")
if err != nil {
return nil, err
}
var objs []interface{}
for obj := iter.Next(); obj != nil; obj = iter.Next() {
objs = append(objs, obj)
}
return objs, nil
}
...
