helm源码分析-storage
storage模块主要用于管理和操作发布的release信息，当我们通过`helm list、helm history等命令查询release信息时，就涉及到存储相关的知识
概述
storage模块定义了存储的基本接口，并提供了不同的实现方式，包括：
- secret：默认方式
- configmap
- memory
- sql：目前只支持postgresql
源码分析
目录结构
storage相关的源码都在 pkg/storage 包下
➜ helm git:(041ce5a2) ✗ tree pkg/storage
pkg/storage
├── driver
│ ├── cfgmaps.go // configmap 这种实现的相关代码
│ ├── cfgmaps_test.go
│ ├── driver.go // 定义了存储的公共接口
│ ├── labels.go
│ ├── labels_test.go
│ ├── memory.go // memory 这种实现的相关代码
│ ├── memory_test.go
│ ├── mock_test.go
│ ├── records.go
│ ├── records_test.go
│ ├── secrets.go // secret 这种实现的相关代码
│ ├── secrets_test.go
│ ├── sql.go // postgresql 这种实现的相关代码
│ ├── sql_test.go
│ └── util.go
├── storage.go // 存储接口定义
└── storage_test.godriver
代码语言：javascript复制type Driver interface {
Creator
Updator
Deletor
Queryor
Name() string
}Creator
代码语言：javascript复制type Creator interface {
Create(key string, rls *rspb.Release) error
}Updator
代码语言：javascript复制type Updator interface {
Update(key string, rls *rspb.Release) error
}Deletor
代码语言：javascript复制type Deletor interface {
Delete(key string) (*rspb.Release, error)
}Quaryor
代码语言：javascript复制type Queryor interface {
Get(key string) (*rspb.Release, error)
List(filter func(*rspb.Release) bool) ([]*rspb.Release, error)
Query(labels map[string]string) ([]*rspb.Release, error)
}storage
代码语言：javascript复制type Storage struct {
driver.Driver
// MaxHistory specifies the maximum number of historical releases that will
// be retained, including the most recent release. Values of 0 or less are
// ignored (meaning no limits are imposed).
MaxHistory int
Log func(string, ...interface{})
}secret实现
secret的实现主要是将Release对象做base64编码，并保存到k8s的secret这种资源类型中
pkg/storage/driver/secret.go
代码语言：javascript复制type Secrets struct {
impl corev1.SecretInterface
Log func(string, ...interface{})
}Get
代码语言：javascript复制func (secrets *Secrets) Get(key string) (*rspb.Release, error) {
// 通过给定的key，调用k8s的api查询seecret
obj, err := secrets.impl.Get(context.Background(), key, metav1.GetOptions{})
...
// 找到secret中的数据，并做base64解码，就得到了Release对象
r, err := decodeRelease(string(obj.Data["release"]))
return r, errors.Wrapf(err, "get: failed to decode data %q", key)
}List
代码语言：javascript复制func (secrets *Secrets) sList(filter func(*rspb.Release) bool) ([]*rspb.Release, error) {
// 构造Label选择器，选择owner=helm的label
lsel := kblabels.Set{"owner": "helm"}.AsSelector()
opts := metav1.ListOptions{LabelSelector: lsel.String()}
// 调用k8s的原生api，获取所有owner=helm的secret
list, err := secrets.impl.List(context.Background(), opts)
...
// 结果处理
for _, item := range list.Items {
// 获取每个secret中的数据，转换为Release对象
rls, err := decodeRelease(string(item.Data["release"]))
...
}
return results, nil
}Create
代码语言：javascript复制func (secrets *Secrets) Create(key string, rls *rspb.Release) error {
// set labels for secrets object meta data
var lbs labels
// 初始化label信息
lbs.init()
lbs.set("createdAt", strconv.Itoa(int(time.Now().Unix())))
// 创建一个持有release信息的secret
obj, err := newSecretsObject(key, rls, lbs)
...
// god k8s的接口，创建secret这个资源
if _, err := secrets.impl.Create(context.Background(), obj, metav1.CreateOptions{}); err != nil {
...
}
return nil
}初始化secret对象
代码语言：javascript复制func newSecretsObject(key string, rls *rspb.Release, lbs labels) (*v1.Secret, error) {
const owner = "helm"
// release做base64编码
s, err := encodeRelease(rls)
...
// 设置label，包括：name、owner、status、version
// 其中owner和name是后续查询release的关键参数
lbs.set("name", rls.Name)
lbs.set("owner", owner)
lbs.set("status", rls.Info.Status.String())
lbs.set("version", strconv.Itoa(rls.Version))
// 添加了Type这个字段，并设置data
return &v1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: key,
Labels: lbs.toMap(),
},
Type: "helm.sh/release.v1",
Data: map[string][]byte{"release": []byte(s)},
}, nil
}Quary
代码语言：javascript复制func (secrets *Secrets) Query(labels map[string]string) ([]*rspb.Release, error) {
ls := kblabels.Set{}
for k, v := range labels {
if errs := validation.IsValidLabelValue(v); len(errs) != 0 {
return nil, errors.Errorf("invalid label value: %q: %s", v, strings.Join(errs, "; "))
}
ls[k] = v
}
opts := metav1.ListOptions{LabelSelector: ls.AsSelector().String()}
list, err := secrets.impl.List(context.Background(), opts)
if err != nil {
return nil, errors.Wrap(err, "query: failed to query with labels")
}
if len(list.Items) == 0 {
return nil, ErrReleaseNotFound
}
var results []*rspb.Release
for _, item := range list.Items {
rls, err := decodeRelease(string(item.Data["release"]))
if err != nil {
secrets.Log("query: failed to decode release: %s", err)
continue
}
results = append(results, rls)
}
return results, nil
}Update
代码语言：javascript复制func (secrets *Secrets) Update(key string, rls *rspb.Release) error {
...
// 设置资源更新时间
lbs.set("modifiedAt", strconv.Itoa(int(time.Now().Unix())))
// 创建一个新的secret资源对象
obj, err := newSecretsObject(key, rls, lbs)
if err != nil {
return errors.Wrapf(err, "update: failed to encode release %q", rls.Name)
}
// 调用k8s的接口，更新secret资源
_, err = secrets.impl.Update(context.Background(), obj, metav1.UpdateOptions{})
return errors.Wrap(err, "update: failed to update")
}Delete
代码语言：javascript复制func (secrets *Secrets) Delete(key string) (rls *rspb.Release, err error) {
// 先获取对应的资源信息
if rls, err = secrets.Get(key); err != nil {
return nil, err
}
// 调用k8s的api删除对应的secret
err = secrets.impl.Delete(context.Background(), key, metav1.DeleteOptions{})
return rls, err
}configmap实现
configmap的实现和secret类似，不再重复分析了。唯一的区别就是，操作的资源是k8s中的configmap，而不是secret
sql实现
sql实现是将release信息存储到postgresql数据库中，
表结构信息
其中涉及到的表结构信息如下
pkg/storage/driver/sql.go
代码语言：javascript复制// 支持的数据库：postgresql
const postgreSQLDialect = "postgres"
// SQLDriverName is the string name of this driver.
const SQLDriverName = "SQL"
// 存储数据的表明
const sqlReleaseTableName = "releases_v1"
// 存储数据的字段名
const (
sqlReleaseTableKeyColumn = "key"
sqlReleaseTableTypeColumn = "type"
sqlReleaseTableBodyColumn = "body"
sqlReleaseTableNameColumn = "name"
sqlReleaseTableNamespaceColumn = "namespace"
sqlReleaseTableVersionColumn = "version"
sqlReleaseTableStatusColumn = "status"
sqlReleaseTableOwnerColumn = "owner"
sqlReleaseTableCreatedAtColumn = "createdAt"
sqlReleaseTableModifiedAtColumn = "modifiedAt"
)创建表结构
代码语言：javascript复制func NewSQL(connectionString string, logger func(string, ...interface{}), namespace string) (*SQL, error) {
// 连接数据库
db, err := sqlx.Connect(postgreSQLDialect, connectionString)
if err != nil {
return nil, err
}
driver := &SQL{
db: db,
Log: logger,
statementBuilder: sq.StatementBuilder.PlaceholderFormat(sq.Dollar),
}
// 表结构初始化
if err := driver.ensureDBSetup(); err != nil {
return nil, err
}
driver.namespace = namespace
return driver, nil
}
func (s *SQL) ensureDBSetup() error {
...
}Get
代码语言：javascript复制func (s *SQL) Get(key string) (*rspb.Release, error) {
var record SQLReleaseWrapper
// 拼接sql语句：select body from releases_v1 where key = ? and namespace = ?
qb := s.statementBuilder.
Select(sqlReleaseTableBodyColumn).
From(sqlReleaseTableName).
Where(sq.Eq{sqlReleaseTableKeyColumn: key}).
Where(sq.Eq{sqlReleaseTableNamespaceColumn: s.namespace})
query, args, err := qb.ToSql()
...
// 获取记录
if err := s.db.Get(&record, query, args...); err != nil {
s.Log("got SQL error when getting release %s: %v", key, err)
return nil, ErrReleaseNotFound
}
// base64解码
release, err := decodeRelease(record.Body)
...
return release, nil
}List
对应的sql语句为：
代码语言：javascript复制select body from releases_v1 where owner = helmCreate
对应的sql语句为：
代码语言：javascript复制insert into release_v1(key1, key2, ...) values(val1, val2, ...)Quary
对应的sql语句为：
代码语言：javascript复制select body from release_v1 where label1 = value1 and ...Update
对应的sql语句为：
代码语言：javascript复制update body set key1=val1 where key = ? and namespace = ?Delete
对应的sql语句为：
代码语言：javascript复制delete releases_v1 where key = ? and namespace = ?memory实现
memory是基于内存实现的存储，数据存放在Memeory这个对象中。
该对象按照命名空间、名称的层级划分，memory的存储驱动就是对这个数据结构的增删改查，本文只分析Get操作。
Memory对象
代码语言：javascript复制type Memory struct {
sync.RWMutex
namespace string
// key是命名空间
cache map[string]memReleases
}
// key是release名称，value是release对应的多个版本数据
type memReleases map[string]records
type records []*recordGet
代码语言：javascript复制func (mem *Memory) Get(key string) (*rspb.Release, error) {
...
keyWithoutPrefix := strings.TrimPrefix(key, "sh.helm.release.v1.")
switch elems := strings.Split(keyWithoutPrefix, ".v"); len(elems) {
case 2:
...
// 根据namespace、name的层级结构定位到数据
if recs, ok := mem.cache[mem.namespace][name]; ok {
if r := recs.Get(key); r != nil {
return r.rls, nil
}
}
return nil, ErrReleaseNotFound
default:
return nil, ErrInvalidKey
}
}
