helm源码分析-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.go

type Driver interface {
    Creator
    Updator
    Deletor
    Queryor
    Name() string
}

type Creator interface {
    Create(key string, rls *rspb.Release) error
}

type Updator interface {
    Update(key string, rls *rspb.Release) error
}

type Deletor interface {
    Delete(key string) (*rspb.Release, error)
}

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)
}

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{})
}

type Secrets struct {
    impl corev1.SecretInterface
    Log  func(string, ...interface{})
}

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)
}

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
}

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
}

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
}

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
}

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")
}

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
}

// 支持的数据库：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"
)

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 {
  ...
}

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
}

select body from releases_v1 where owner = helm

insert into release_v1(key1, key2, ...) values(val1, val2, ...)

select body from release_v1 where label1 = value1 and ...

update body set key1=val1 where key = ? and namespace = ?

delete releases_v1 where key = ? and namespace = ?

type Memory struct {
    sync.RWMutex
    namespace string
    // key是命名空间
    cache map[string]memReleases
}

// key是release名称，value是release对应的多个版本数据
type memReleases map[string]records

type records []*record

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
    }
}