admin/kubesphere
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #4250 from xyz-li/app_store

Browse Source 
Dynamically load helm app into app-store
This commit is contained in:
KubeSphere CI Bot
2021-09-29 10:14:54 +08:00
committed by GitHub
parent 60af01c10d 09fc2867c4
commit 22ae83a431
16 changed files with 422 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/constants/constants.go
View File

@@ -71,6 +71,8 @@ const (
OpenpitrixAttachmentTag = "Attachment"
OpenpitrixRepositoryTag = "Repository"
OpenpitrixManagementTag = "App Management"
// HelmRepoMinSyncPeriod min sync period in seconds
HelmRepoMinSyncPeriod = 180
CleanupDanglingAppOngoing = "ongoing"
CleanupDanglingAppDone = "done"

17
pkg/controller/openpitrix/helmrepo/helm_repo_controller.go
View File

@@ -35,6 +35,8 @@ import (
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"
"kubesphere.io/kubesphere/pkg/constants"
"kubesphere.io/api/application/v1alpha1"
"kubesphere.io/kubesphere/pkg/simple/client/openpitrix/helmrepoindex"
@@ -42,9 +44,6 @@ import (
)
const (
// min sync period in seconds
MinSyncPeriod = 180
MinRetryDuration = 60
MaxRetryDuration = 600
HelmRepoSyncStateLen = 10
@@ -156,8 +155,8 @@ func (r *ReconcileHelmRepo) Reconcile(ctx context.Context, request reconcile.Req
copyInstance := instance.DeepCopy()
if copyInstance.Spec.SyncPeriod != 0 && copyInstance.Spec.SyncPeriod < MinSyncPeriod {
copyInstance.Spec.SyncPeriod = MinSyncPeriod
if copyInstance.Spec.SyncPeriod != 0 {
copyInstance.Spec.SyncPeriod = int(math.Max(float64(copyInstance.Spec.SyncPeriod), constants.HelmRepoMinSyncPeriod))
}
retryAfter := 0
@@ -197,7 +196,7 @@ func (r *ReconcileHelmRepo) Reconcile(ctx context.Context, request reconcile.Req
RequeueAfter: MinRetryDuration * time.Second,
}, err
} else {
retryAfter = MinSyncPeriod
retryAfter = constants.HelmRepoMinSyncPeriod
if syncErr == nil {
retryAfter = copyInstance.Spec.SyncPeriod
}
@@ -256,9 +255,7 @@ func needReSyncNow(instance *v1alpha1.HelmRepo) (syncNow bool, after int) {
} else {
period = instance.Spec.SyncPeriod
if period != 0 {
if period < MinSyncPeriod {
period = MinSyncPeriod
}
period = int(math.Max(float64(instance.Spec.SyncPeriod), constants.HelmRepoMinSyncPeriod))
if now.After(state.SyncTime.Add(time.Duration(period) * time.Second)) {
return true, 0
}
@@ -296,7 +293,7 @@ func (r *ReconcileHelmRepo) syncRepo(instance *v1alpha1.HelmRepo) error {
}
// 2. merge new index with old index which is stored in crd
savedIndex := helmrepoindex.MergeRepoIndex(index, existsSavedIndex)
savedIndex := helmrepoindex.MergeRepoIndex(instance, index, existsSavedIndex)
// 3. save index in crd
data, err := savedIndex.Bytes()

28
pkg/kapis/openpitrix/v1/handler.go
View File

@@ -17,9 +17,11 @@ import (
"encoding/json"
"fmt"
"io/ioutil"
"math"
"net/url"
"strconv"
"strings"
"time"
restful "github.com/emicklei/go-restful"
"google.golang.org/grpc/codes"
@@ -90,6 +92,18 @@ func (h *openpitrixHandler) CreateRepo(req *restful.Request, resp *restful.Respo
// trim credential from url
parsedUrl.User = nil
syncPeriod := 0
// If SyncPeriod is empty, ignore it.
if createRepoRequest.SyncPeriod != "" {
duration, err := time.ParseDuration(createRepoRequest.SyncPeriod)
if err != nil {
api.HandleBadRequest(resp, nil, err)
return
} else if duration > 0 {
syncPeriod = int(math.Max(float64(duration/time.Second), constants.HelmRepoMinSyncPeriod))
}
}
repo := v1alpha1.HelmRepo{
ObjectMeta: metav1.ObjectMeta{
Name: idutils.GetUuid36(v1alpha1.HelmRepoIdPrefix),
@@ -103,11 +117,15 @@ func (h *openpitrixHandler) CreateRepo(req *restful.Request, resp *restful.Respo
Spec: v1alpha1.HelmRepoSpec{
Name: createRepoRequest.Name,
Url: parsedUrl.String(),
SyncPeriod: 0,
SyncPeriod: syncPeriod,
Description: stringutils.ShortenString(createRepoRequest.Description, 512),
},
}
if syncPeriod > 0 {
repo.Annotations[v1alpha1.RepoSyncPeriod] = createRepoRequest.SyncPeriod
}
if strings.HasPrefix(createRepoRequest.URL, "https://") || strings.HasPrefix(createRepoRequest.URL, "http://") {
if userInfo != nil {
repo.Spec.Credential.Username = userInfo.Username()
@@ -322,7 +340,7 @@ func (h *openpitrixHandler) DoAppAction(req *restful.Request, resp *restful.Resp
if err != nil {
klog.Errorln(err)
handleOpenpitrixError(resp, err)
api.HandleError(resp, nil, err)
return
}
@@ -388,7 +406,7 @@ func (h *openpitrixHandler) ModifyApp(req *restful.Request, resp *restful.Respon
if err != nil {
klog.Errorln(err)
handleOpenpitrixError(resp, err)
api.HandleError(resp, nil, err)
return
}
@@ -570,7 +588,7 @@ func (h *openpitrixHandler) ModifyAppVersion(req *restful.Request, resp *restful
if err != nil {
klog.Errorln(err)
handleOpenpitrixError(resp, err)
api.HandleError(resp, nil, err)
return
}
@@ -665,7 +683,7 @@ func (h *openpitrixHandler) DoAppVersionAction(req *restful.Request, resp *restf
if err != nil {
klog.Errorln(err)
handleOpenpitrixError(resp, err)
api.HandleError(resp, nil, err)
return
}

32
pkg/models/openpitrix/applications.go
View File

@@ -175,11 +175,16 @@ func (c *applicationOperator) ValidatePackage(request *ValidatePackageRequest) (
func (c *applicationOperator) DoAppAction(appId string, request *ActionRequest) error {
app, err := c.appLister.Get(appId)
app, err := c.getHelmApplication(appId)
if err != nil {
return err
}
// All the app belonging to a built-in repo have a label `application.kubesphere.io/repo-id`, and the value should be `builtin-stable` or else.
if repoId, exist := app.Labels[constants.ChartRepoIdLabelKey]; exist && repoId != v1alpha1.AppStoreRepoId {
return apierrors.NewForbidden(v1alpha1.Resource(v1alpha1.ResourcePluralHelmApplication), app.Name, errors.New("application is immutable"))
}
var filterState string
switch request.Action {
case ActionSuspend:
@@ -393,12 +398,17 @@ func (c *applicationOperator) ModifyApp(appId string, request *ModifyAppRequest)
return invalidS3Config
}
app, err := c.appLister.Get(appId)
app, err := c.getHelmApplication(appId)
if err != nil {
klog.Error(err)
return err
}
// All the app belonging to a built-in repo have a label `application.kubesphere.io/repo-id`, and the value should be `builtin-stable` or else.
if repoId, exist := app.Labels[constants.ChartRepoIdLabelKey]; exist && repoId != v1alpha1.AppStoreRepoId {
return apierrors.NewForbidden(v1alpha1.Resource(v1alpha1.ResourcePluralHelmApplication), app.Name, errors.New("application is immutable"))
}
appCopy := app.DeepCopy()
// modify category
if request.CategoryID != nil {
@@ -602,16 +612,32 @@ func (c *applicationOperator) listApps(conditions *params.Conditions) (ret []*v1
repoId := conditions.Match[RepoId]
if repoId != "" && repoId != v1alpha1.AppStoreRepoId {
// get helm application from helm repo
if ret, exists := c.cachedRepos.ListApplicationsByRepoId(repoId); !exists {
if ret, exists := c.cachedRepos.ListApplicationsInRepo(repoId); !exists {
klog.Warningf("load repo failed, repo id: %s", repoId)
return nil, loadRepoInfoFailed
} else {
return ret, nil
}
} else if repoId == v1alpha1.AppStoreRepoId {
// List apps in the app-store and built-in repo
if c.backingStoreClient == nil {
return []*v1alpha1.HelmApplication{}, nil
}
ls := map[string]string{}
// We just care about the category label when listing apps in built-in repo.
if conditions.Match[CategoryId] != "" {
ls[constants.CategoryIdLabelKey] = conditions.Match[CategoryId]
}
appInRepo, _ := c.cachedRepos.ListApplicationsInBuiltinRepo(labels.SelectorFromSet(ls))
ret, err = c.appLister.List(labels.SelectorFromSet(buildLabelSelector(conditions)))
ret = append(ret, appInRepo...)
} else {
if c.backingStoreClient == nil {
return []*v1alpha1.HelmApplication{}, nil
}
ret, err = c.appLister.List(labels.SelectorFromSet(buildLabelSelector(conditions)))
}

26
pkg/models/openpitrix/applicationversions.go
View File

@@ -141,7 +141,7 @@ func (c *applicationOperator) DeleteAppVersion(id string) error {
}
func (c *applicationOperator) DescribeAppVersion(id string) (*AppVersion, error) {
version, err := c.versionLister.Get(id)
version, err := c.getAppVersion(id)
if err != nil {
klog.Errorf("get app version [%s] failed, error: %s", id, err)
return nil, err
@@ -152,12 +152,17 @@ func (c *applicationOperator) DescribeAppVersion(id string) (*AppVersion, error)
func (c *applicationOperator) ModifyAppVersion(id string, request *ModifyAppVersionRequest) error {
version, err := c.versionLister.Get(id)
version, err := c.getAppVersion(id)
if err != nil {
klog.Errorf("get app version [%s] failed, error: %s", id, err)
return err
}
// All the app versions belonging to a built-in repo have a label `application.kubesphere.io/repo-id`, and the value should be `builtin-stable` or else.
if repoId, exists := version.Labels[constants.ChartRepoIdLabelKey]; exists && repoId != v1alpha1.AppStoreRepoId {
return apierrors.NewForbidden(v1alpha1.Resource(v1alpha1.ResourcePluralHelmApplicationVersion), version.Name, errors.New("version is immutable"))
}
versionCopy := version.DeepCopy()
spec := &versionCopy.Spec
@@ -355,12 +360,17 @@ func (c *applicationOperator) DoAppVersionAction(versionId string, request *Acti
}
state := v1alpha1.StateDraft
version, err := c.versionLister.Get(versionId)
version, err := c.getAppVersion(versionId)
if err != nil {
klog.Errorf("get app version %s failed, error: %s", versionId, err)
return err
}
// All the app versions belonging to a built-in repo have a label `application.kubesphere.io/repo-id`, and the value should be `builtin-stable` or else.
if repoId, exists := version.Labels[constants.ChartRepoIdLabelKey]; exists && repoId != v1alpha1.AppStoreRepoId {
return apierrors.NewForbidden(v1alpha1.Resource(v1alpha1.ResourcePluralHelmApplicationVersion), version.Name, errors.New("version is immutable"))
}
switch request.Action {
case ActionCancel:
if version.Status.State != v1alpha1.StateSubmitted {
@@ -588,3 +598,13 @@ func (c *applicationOperator) getAppVersionsByAppId(appId string) (ret []*v1alph
return
}
// get app version from repo and helm application
func (c *applicationOperator) getAppVersion(id string) (ret *v1alpha1.HelmApplicationVersion, err error) {
if ver, exists, _ := c.cachedRepos.GetAppVersion(id); exists {
return ver, nil
}
ret, err = c.versionLister.Get(id)
return
}

15
pkg/models/openpitrix/categories.go
View File

@@ -17,6 +17,8 @@ import (
"context"
"sort"
"kubesphere.io/kubesphere/pkg/utils/reposcache"
"kubesphere.io/kubesphere/pkg/apiserver/query"
apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -48,12 +50,14 @@ type CategoryInterface interface {
type categoryOperator struct {
ctgClient typed_v1alpha1.ApplicationV1alpha1Interface
ctgLister listers_v1alpha1.HelmCategoryLister
repoCache reposcache.ReposCache
}
func newCategoryOperator(ksFactory externalversions.SharedInformerFactory, ksClient versioned.Interface) CategoryInterface {
func newCategoryOperator(repoCache reposcache.ReposCache, ksFactory externalversions.SharedInformerFactory, ksClient versioned.Interface) CategoryInterface {
c := &categoryOperator{
ctgClient: ksClient.ApplicationV1alpha1(),
ctgLister: ksFactory.Application().V1alpha1().HelmCategories().Lister(),
repoCache: repoCache,
}
return c
@@ -190,8 +194,15 @@ func (c *categoryOperator) ListCategories(conditions *params.Conditions, orderBy
start, end := (&query.Pagination{Limit: limit, Offset: offset}).GetValidPagination(totalCount)
ctgs = ctgs[start:end]
items := make([]interface{}, 0, len(ctgs))
ctgCountsOfBuiltinRepo := c.repoCache.CopyCategoryCount()
for i := range ctgs {
items = append(items, convertCategory(ctgs[i]))
convertedCtg := convertCategory(ctgs[i])
// The statistic of category for app in etcd is stored in the crd.
// The statistic of category for the app in the built-in repo is stored in the memory.
// So we should calculate these two value then return.
*convertedCtg.AppTotal += ctgCountsOfBuiltinRepo[convertedCtg.CategoryID]
items = append(items, convertedCtg)
}
return &models.PageableResponse{Items: items, TotalCount: totalCount}, nil

4
pkg/models/openpitrix/category_test.go
View File

@@ -20,6 +20,8 @@ import (
"context"
"testing"
"kubesphere.io/kubesphere/pkg/utils/reposcache"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
fakek8s "k8s.io/client-go/kubernetes/fake"
"k8s.io/klog"
@@ -82,5 +84,5 @@ func prepareCategoryOperator() CategoryInterface {
k8sClient = fakek8s.NewSimpleClientset()
fakeInformerFactory = informers.NewInformerFactories(k8sClient, ksClient, nil, nil, nil, nil)
return newCategoryOperator(fakeInformerFactory.KubeSphereSharedInformerFactory(), ksClient)
return newCategoryOperator(reposcache.NewReposCache(), fakeInformerFactory.KubeSphereSharedInformerFactory(), ksClient)
}

23
pkg/models/openpitrix/interface.go
View File

@@ -56,7 +56,6 @@ func init() {
}
func NewOpenpitrixOperator(ksInformers ks_informers.InformerFactory, ksClient versioned.Interface, s3Client s3.Interface) Interface {
once.Do(func() {
klog.Infof("start helm repo informer")
helmReposInformer = ksInformers.KubeSphereSharedInformerFactory().Application().V1alpha1().HelmRepos().Informer()
@@ -66,16 +65,28 @@ func NewOpenpitrixOperator(ksInformers ks_informers.InformerFactory, ksClient ve
cachedReposData.AddRepo(r)
},
UpdateFunc: func(oldObj, newObj interface{}) {
oldR := oldObj.(*v1alpha1.HelmRepo)
cachedReposData.DeleteRepo(oldR)
r := newObj.(*v1alpha1.HelmRepo)
cachedReposData.AddRepo(r)
oldRepo := oldObj.(*v1alpha1.HelmRepo)
newRepo := newObj.(*v1alpha1.HelmRepo)
cachedReposData.UpdateRepo(oldRepo, newRepo)
},
DeleteFunc: func(obj interface{}) {
r := obj.(*v1alpha1.HelmRepo)
cachedReposData.DeleteRepo(r)
},
})
ctgInformer := ksInformers.KubeSphereSharedInformerFactory().Application().V1alpha1().HelmCategories().Informer()
ctgInformer.AddIndexers(map[string]cache.IndexFunc{
reposcache.CategoryIndexer: func(obj interface{}) ([]string, error) {
ctg, _ := obj.(*v1alpha1.HelmCategory)
return []string{ctg.Spec.Name}, nil
},
})
indexer := ctgInformer.GetIndexer()
cachedReposData.SetCategoryIndexer(indexer)
go ctgInformer.Run(wait.NeverStop)
go helmReposInformer.Run(wait.NeverStop)
})
@@ -84,6 +95,6 @@ func NewOpenpitrixOperator(ksInformers ks_informers.InformerFactory, ksClient ve
ApplicationInterface: newApplicationOperator(cachedReposData, ksInformers.KubeSphereSharedInformerFactory(), ksClient, s3Client),
RepoInterface: newRepoOperator(cachedReposData, ksInformers.KubeSphereSharedInformerFactory(), ksClient),
ReleaseInterface: newReleaseOperator(cachedReposData, ksInformers.KubernetesSharedInformerFactory(), ksInformers.KubeSphereSharedInformerFactory(), ksClient),
CategoryInterface: newCategoryOperator(ksInformers.KubeSphereSharedInformerFactory(), ksClient),
CategoryInterface: newCategoryOperator(cachedReposData, ksInformers.KubeSphereSharedInformerFactory(), ksClient),
}
}

27
pkg/models/openpitrix/repos.go
View File

@@ -19,6 +19,7 @@ import (
"net/url"
"sort"
"strings"
"time"
"kubesphere.io/kubesphere/pkg/apiserver/query"
@@ -162,6 +163,32 @@ func (c *repoOperator) ModifyRepo(id string, request *ModifyRepoRequest) error {
repoCopy.Spec.Description = stringutils.ShortenString(*request.Description, DescriptionLen)
}
if repoCopy.Annotations == nil {
repoCopy.Annotations = map[string]string{}
}
if request.SyncPeriod != nil {
syncPeriod := 0
if *request.SyncPeriod == "" {
// disable auto sync
syncPeriod = 0
} else {
if duration, err := time.ParseDuration(*request.SyncPeriod); err != nil {
return err
} else {
syncPeriod = int(duration / time.Second)
}
}
if syncPeriod == 0 {
// disable auto sync
repoCopy.Spec.SyncPeriod = 0
delete(repoCopy.Annotations, v1alpha1.RepoSyncPeriod)
} else {
repoCopy.Spec.SyncPeriod = syncPeriod
repoCopy.Annotations[v1alpha1.RepoSyncPeriod] = *request.SyncPeriod
}
}
// modify name of the repo
if request.Name != nil && len(*request.Name) > 0 && *request.Name != repoCopy.Spec.Name {
items, err := c.repoLister.List(labels.SelectorFromSet(map[string]string{constants.WorkspaceLabelKey: repo.GetWorkspace()}))

10
pkg/models/openpitrix/types.go
View File

@@ -585,6 +585,11 @@ type CreateRepoRequest struct {
// required, runtime provider eg.[qingcloud|aliyun|aws|kubernetes]
Providers []string `json:"providers"`
// min sync period to sync helm repo, a duration string is a sequence of
// decimal numbers, each with optional fraction and a unit suffix,
// such as "180s", "2h" or "45m".
SyncPeriod string `json:"sync_period"`
// repository type
Type string `json:"type,omitempty"`
@@ -612,6 +617,9 @@ type ModifyRepoRequest struct {
Workspace *string `json:"workspace,omitempty"`
// min sync period to sync helm repo
SyncPeriod *string `json:"sync_period"`
// repository name
Name *string `json:"name,omitempty"`
@@ -716,6 +724,8 @@ type Repo struct {
// visibility.eg:[public|private]
Visibility string `json:"visibility,omitempty"`
SyncPeriod string `json:"sync_period,omitempty"`
}
type CreateRepoResponse struct {

15
pkg/models/openpitrix/utils.go
View File

@@ -401,6 +401,18 @@ func convertAppVersion(in *v1alpha1.HelmApplicationVersion) *AppVersion {
out.Icon = in.Spec.Icon
}
// The field Maintainers and Sources were a string field, so I encode the helm field's maintainers and sources,
// which are array, to string.
if len(in.Spec.Maintainers) > 0 {
maintainers, _ := json.Marshal(in.Spec.Maintainers)
out.Maintainers = string(maintainers)
}
if len(in.Spec.Sources) > 0 {
source, _ := json.Marshal(in.Spec.Sources)
out.Sources = string(source)
}
out.Status = in.State()
out.Owner = in.GetCreator()
out.Name = in.GetVersionName()
@@ -427,6 +439,7 @@ func convertRepo(in *v1alpha1.HelmRepo) *Repo {
cred, _ := json.Marshal(in.Spec.Credential)
out.Credential = string(cred)
out.SyncPeriod = in.Annotations[v1alpha1.RepoSyncPeriod]
out.URL = in.Spec.Url
return &out
@@ -564,6 +577,8 @@ func buildApplicationVersion(app *v1alpha1.HelmApplication, chrt helmrepoindex.V
Icon: chrt.GetIcon(),
Home: chrt.GetHome(),
Description: stringutils.ShortenString(chrt.GetDescription(), v1alpha1.MsgLen),
Sources: chrt.GetRawSources(),
Maintainers: chrt.GetRawMaintainers(),
},
Created: &t,
// set data to nil before save app version to etcd

8
pkg/simple/client/openpitrix/helmrepoindex/interface.go
View File

@@ -16,7 +16,11 @@ limitations under the License.
package helmrepoindex
import "time"
import (
"time"
"kubesphere.io/api/application/v1alpha1"
)
type VersionInterface interface {
GetName() string
@@ -28,8 +32,10 @@ type VersionInterface interface {
GetIcon() string
GetHome() string
GetSources() string
GetRawSources() []string
GetKeywords() string
GetMaintainers() string
GetRawMaintainers() []*v1alpha1.Maintainer
GetScreenshots() string
GetPackageName() string
GetCreateTime() time.Time

19
pkg/simple/client/openpitrix/helmrepoindex/load_package.go
View File

@@ -23,6 +23,8 @@ import (
"strings"
"time"
"kubesphere.io/api/application/v1alpha1"
"helm.sh/helm/v3/pkg/repo"
"k8s.io/klog"
@@ -64,10 +66,27 @@ func (h HelmVersionWrapper) GetSources() string {
return string(s)
}
func (h HelmVersionWrapper) GetRawSources() []string {
return h.Sources
}
func (h HelmVersionWrapper) GetKeywords() string {
return strings.Join(h.ChartVersion.Keywords, ",")
}
func (h HelmVersionWrapper) GetRawMaintainers() []*v1alpha1.Maintainer {
mt := make([]*v1alpha1.Maintainer, 0, len(h.Maintainers))
for _, value := range h.Maintainers {
mt = append(mt, &v1alpha1.Maintainer{
URL: value.URL,
Name: value.Name,
Email: value.Email,
})
}
return mt
}
func (h HelmVersionWrapper) GetMaintainers() string {
if len(h.ChartVersion.Maintainers) == 0 {
return ""

68
pkg/simple/client/openpitrix/helmrepoindex/repo_index.go
View File

@@ -78,8 +78,8 @@ func loadIndex(data []byte) (*helmrepo.IndexFile, error) {
var empty = struct{}{}
// merge new index with index from crd
func MergeRepoIndex(index *helmrepo.IndexFile, existsSavedIndex *SavedIndex) *SavedIndex {
// MergeRepoIndex merge new index with index from crd
func MergeRepoIndex(repo *v1alpha1.HelmRepo, index *helmrepo.IndexFile, existsSavedIndex *SavedIndex) *SavedIndex {
saved := &SavedIndex{}
if index == nil {
return existsSavedIndex
@@ -102,20 +102,37 @@ func MergeRepoIndex(index *helmrepo.IndexFile, existsSavedIndex *SavedIndex) *Sa
// add new applications
if application, exists := saved.Applications[name]; !exists {
application = &Application{
Name: name,
ApplicationId: idutils.GetUuid36(v1alpha1.HelmApplicationIdPrefix),
Description: versions[0].Description,
Icon: versions[0].Icon,
Name: name,
Description: versions[0].Description,
Icon: versions[0].Icon,
Created: time.Now(),
}
// The app id will be added to the labels of the helm release.
// But the apps in the repos which are created by the user may contain malformed text, so we generate a random name for them.
// The apps in the system repo have been audited by the admin, so the name of the charts should not include malformed text.
// Then we can add the name string to the labels of the k8s object.
if IsBuiltInRepo(repo.Name) {
application.ApplicationId = fmt.Sprintf("%s%s-%s", v1alpha1.HelmApplicationIdPrefix, repo.Name, name)
} else {
application.ApplicationId = idutils.GetUuid36(v1alpha1.HelmApplicationIdPrefix)
}
charts := make([]*ChartVersion, 0, len(versions))
for ind := range versions {
chart := &ChartVersion{
ApplicationId: application.ApplicationId,
ApplicationVersionId: idutils.GetUuid36(v1alpha1.HelmApplicationVersionIdPrefix),
ChartVersion: *versions[ind],
ApplicationId: application.ApplicationId,
ChartVersion: *versions[ind],
}
chart.ApplicationVersionId = generateAppVersionId(repo, versions[ind].Name, versions[ind].Version)
charts = append(charts, chart)
// Use the creation time of the oldest chart as the creation time of the app.
if versions[ind].Created.Before(application.Created) {
application.Created = versions[ind].Created
}
}
application.Charts = charts
@@ -132,10 +149,11 @@ func MergeRepoIndex(index *helmrepo.IndexFile, existsSavedIndex *SavedIndex) *Sa
// add new chart version
if _, exists := savedChartVersion[ver.Version]; !exists {
chart := &ChartVersion{
ApplicationId: application.ApplicationId,
ApplicationVersionId: idutils.GetUuid36(v1alpha1.HelmApplicationVersionIdPrefix),
ChartVersion: *ver,
ApplicationId: application.ApplicationId,
ChartVersion: *ver,
}
chart.ApplicationVersionId = generateAppVersionId(repo, ver.Name, ver.Version)
charts = append(charts, chart)
}
newVersion[ver.Version] = empty
@@ -204,6 +222,26 @@ func (i *SavedIndex) GetApplicationVersion(appId, versionId string) *v1alpha1.He
return nil
}
// The app version id will be added to the labels of the helm release.
// But the apps in the repos which are created by the user may contain malformed text, so we generate a random name for them.
// The apps in the system repo have been audited by the admin, so the name of the charts should not include malformed text.
// Then we can add the name string to the labels of the k8s object.
func generateAppVersionId(repo *v1alpha1.HelmRepo, chartName, version string) string {
if IsBuiltInRepo(repo.Name) {
return fmt.Sprintf("%s%s-%s-%s", v1alpha1.HelmApplicationIdPrefix, repo.Name, chartName, version)
} else {
return idutils.GetUuid36(v1alpha1.HelmApplicationVersionIdPrefix)
}
}
// IsBuiltInRepo checks whether a repo is a built-in repo.
// All the built-in repos are located in the workspace system-workspace and the name starts with 'built-in'
// to differentiate from the repos created by the user.
func IsBuiltInRepo(repoName string) bool {
return strings.HasPrefix(repoName, v1alpha1.BuiltinRepoPrefix)
}
type SavedIndex struct {
APIVersion string `json:"apiVersion"`
Generated time.Time `json:"generated"`
@@ -290,7 +328,7 @@ type Application struct {
// application status
Status string `json:"status"`
// The URL to an icon file.
Icon string `json:"icon,omitempty"`
Charts []*ChartVersion `json:"charts"`
Icon string `json:"icon,omitempty"`
Created time.Time `json:"created"`
Charts []*ChartVersion `json:"charts"`
}

240
pkg/utils/reposcache/repo_cahes.go
View File

@@ -19,7 +19,6 @@ package reposcache
import (
"context"
"encoding/base64"
"errors"
"fmt"
"io/ioutil"
@@ -28,7 +27,11 @@ import (
"strings"
"sync"
"k8s.io/client-go/tools/cache"
"github.com/Masterminds/semver/v3"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/klog"
"kubesphere.io/api/application/v1alpha1"
@@ -37,65 +40,89 @@ import (
"kubesphere.io/kubesphere/pkg/simple/client/openpitrix/helmrepoindex"
)
const (
CategoryIndexer = "category_indexer"
CategoryAnnotationKey = "app.kubesphere.io/category"
)
var WorkDir string
func NewReposCache() ReposCache {
return &cachedRepos{
chartsInRepo: map[workspace]map[string]int{},
repos: map[string]*v1alpha1.HelmRepo{},
apps: map[string]*v1alpha1.HelmApplication{},
versions: map[string]*v1alpha1.HelmApplicationVersion{},
repoCtgCounts: map[string]map[string]int{},
chartsInRepo: map[workspace]map[string]int{},
repos: map[string]*v1alpha1.HelmRepo{},
apps: map[string]*v1alpha1.HelmApplication{},
versions: map[string]*v1alpha1.HelmApplicationVersion{},
builtinCategoryCounts: map[string]int{},
}
}
type ReposCache interface {
AddRepo(repo *v1alpha1.HelmRepo) error
DeleteRepo(repo *v1alpha1.HelmRepo) error
UpdateRepo(old, new *v1alpha1.HelmRepo) error
GetApplication(string) (*v1alpha1.HelmApplication, bool)
GetAppVersion(string) (*v1alpha1.HelmApplicationVersion, bool, error)
GetAppVersionWithData(string) (*v1alpha1.HelmApplicationVersion, bool, error)
ListAppVersionsByAppId(appId string) (ret []*v1alpha1.HelmApplicationVersion, exists bool)
ListApplicationsByRepoId(repoId string) (ret []*v1alpha1.HelmApplication, exists bool)
ListApplicationsInRepo(repoId string) (ret []*v1alpha1.HelmApplication, exists bool)
ListApplicationsInBuiltinRepo(selector labels.Selector) (ret []*v1alpha1.HelmApplication, exists bool)
SetCategoryIndexer(indexer cache.Indexer)
CopyCategoryCount() map[string]int
}
type workspace string
type cachedRepos struct {
sync.RWMutex
chartsInRepo map[workspace]map[string]int
repoCtgCounts map[string]map[string]int
chartsInRepo map[workspace]map[string]int
// builtinCategoryCounts saves the count of every category in the built-in repo.
builtinCategoryCounts map[string]int
repos map[string]*v1alpha1.HelmRepo
apps map[string]*v1alpha1.HelmApplication
versions map[string]*v1alpha1.HelmApplicationVersion
// indexerOfHelmCtg is the indexer of HelmCategory, used to query the category id from category name.
indexerOfHelmCtg cache.Indexer
}
func (c *cachedRepos) deleteRepo(repo *v1alpha1.HelmRepo) {
if len(repo.Status.Data) == 0 {
return
}
index, err := helmrepoindex.ByteArrayToSavedIndex([]byte(repo.Status.Data))
if err != nil {
klog.Errorf("json unmarshal repo %s failed, error: %s", repo.Name, err)
return
}
klog.V(4).Infof("delete repo %s from cache", repo.Name)
c.Lock()
defer c.Unlock()
klog.V(2).Infof("delete repo %s from cache", repo.Name)
repoId := repo.GetHelmRepoId()
ws := workspace(repo.GetWorkspace())
if _, exists := c.chartsInRepo[ws]; exists {
delete(c.chartsInRepo[ws], repoId)
}
delete(c.repoCtgCounts, repoId)
delete(c.repos, repoId)
for _, app := range index.Applications {
if _, exists := c.apps[app.ApplicationId]; !exists {
continue
}
if helmrepoindex.IsBuiltInRepo(repo.Name) {
ctgId := c.apps[app.ApplicationId].Labels[constants.CategoryIdLabelKey]
if ctgId != "" {
c.builtinCategoryCounts[ctgId] -= 1
}
}
delete(c.apps, app.ApplicationId)
for _, ver := range app.Charts {
delete(c.versions, ver.ApplicationVersionId)
@@ -118,10 +145,47 @@ func loadBuiltinChartData(name, version string) ([]byte, error) {
}
func (c *cachedRepos) DeleteRepo(repo *v1alpha1.HelmRepo) error {
c.Lock()
defer c.Unlock()
c.deleteRepo(repo)
return nil
}
// CopyCategoryCount copies the internal map to avoid `concurrent map iteration and map write`.
func (c *cachedRepos) CopyCategoryCount() map[string]int {
c.RLock()
defer c.RUnlock()
ret := make(map[string]int, len(c.builtinCategoryCounts))
for k, v := range c.builtinCategoryCounts {
ret[k] = v
}
return ret
}
func (c *cachedRepos) SetCategoryIndexer(indexer cache.Indexer) {
c.Lock()
c.indexerOfHelmCtg = indexer
c.Unlock()
}
// translateCategoryNameToId translate a category-name to a category-id.
// The caller should hold the lock
func (c *cachedRepos) translateCategoryNameToId(ctgName string) string {
if c.indexerOfHelmCtg == nil || ctgName == "" {
return v1alpha1.UncategorizedId
}
if items, err := c.indexerOfHelmCtg.ByIndex(CategoryIndexer, ctgName); len(items) == 0 || err != nil {
return v1alpha1.UncategorizedId
} else {
obj, _ := items[0].(*v1alpha1.HelmCategory)
return obj.Name
}
}
func (c *cachedRepos) GetApplication(appId string) (app *v1alpha1.HelmApplication, exists bool) {
c.RLock()
defer c.RUnlock()
@@ -131,11 +195,24 @@ func (c *cachedRepos) GetApplication(appId string) (app *v1alpha1.HelmApplicatio
return
}
func (c *cachedRepos) UpdateRepo(old, new *v1alpha1.HelmRepo) error {
if old.Status.Data == new.Status.Data {
return nil
}
c.Lock()
defer c.Unlock()
c.deleteRepo(old)
return c.addRepo(new, false)
}
func (c *cachedRepos) AddRepo(repo *v1alpha1.HelmRepo) error {
c.Lock()
defer c.Unlock()
return c.addRepo(repo, false)
}
//Add new Repo to cachedRepos
// Add a new Repo to cachedRepos
func (c *cachedRepos) addRepo(repo *v1alpha1.HelmRepo, builtin bool) error {
if len(repo.Status.Data) == 0 {
return nil
@@ -146,10 +223,7 @@ func (c *cachedRepos) addRepo(repo *v1alpha1.HelmRepo, builtin bool) error {
return err
}
klog.V(4).Infof("add repo %s to cache", repo.Name)
c.Lock()
defer c.Unlock()
klog.V(2).Infof("add repo %s to cache", repo.Name)
ws := workspace(repo.GetWorkspace())
if _, exists := c.chartsInRepo[ws]; !exists {
@@ -158,29 +232,27 @@ func (c *cachedRepos) addRepo(repo *v1alpha1.HelmRepo, builtin bool) error {
repoId := repo.GetHelmRepoId()
c.repos[repoId] = repo
//c.repoCtgCounts[repo.GetHelmRepoId()] = make(map[string]int)
if _, exists := c.repoCtgCounts[repoId]; !exists {
c.repoCtgCounts[repoId] = map[string]int{}
}
var appName string
chartsCount := 0
for key, app := range index.Applications {
if builtin {
appName = v1alpha1.HelmApplicationIdPrefix + app.Name
} else {
appName = app.ApplicationId
appName = app.ApplicationId
appLabels := make(map[string]string)
if helmrepoindex.IsBuiltInRepo(repo.Name) {
appLabels[constants.WorkspaceLabelKey] = "system-workspace"
}
HelmApp := v1alpha1.HelmApplication{
appLabels[constants.ChartRepoIdLabelKey] = repoId
helmApp := v1alpha1.HelmApplication{
ObjectMeta: metav1.ObjectMeta{
Name: appName,
Annotations: map[string]string{
constants.CreatorAnnotationKey: repo.GetCreator(),
},
Labels: map[string]string{
constants.ChartRepoIdLabelKey: repo.GetHelmRepoId(),
},
Labels: appLabels,
CreationTimestamp: metav1.Time{Time: app.Created},
},
Spec: v1alpha1.HelmApplicationSpec{
Name: key,
@@ -191,25 +263,18 @@ func (c *cachedRepos) addRepo(repo *v1alpha1.HelmRepo, builtin bool) error {
State: v1alpha1.StateActive,
},
}
c.apps[app.ApplicationId] = &HelmApp
c.apps[app.ApplicationId] = &helmApp
var ctg, appVerName string
var chartData []byte
for _, ver := range app.Charts {
chartsCount += 1
if ver.Annotations != nil && ver.Annotations["category"] != "" {
ctg = ver.Annotations["category"]
}
if builtin {
appVerName = base64.StdEncoding.EncodeToString([]byte(ver.Name + ver.Version))
chartData, err = loadBuiltinChartData(ver.Name, ver.Version)
if err != nil {
return err
}
} else {
appVerName = ver.ApplicationVersionId
}
var latestVersionName string
var latestSemver *semver.Version
// build all the versions of this app
for _, chartVersion := range app.Charts {
chartsCount += 1
hvw := helmrepoindex.HelmVersionWrapper{ChartVersion: &chartVersion.ChartVersion}
appVerName = chartVersion.ApplicationVersionId
version := &v1alpha1.HelmApplicationVersion{
ObjectMeta: metav1.ObjectMeta{
Name: appVerName,
@@ -218,42 +283,64 @@ func (c *cachedRepos) addRepo(repo *v1alpha1.HelmRepo, builtin bool) error {
constants.ChartApplicationIdLabelKey: appName,
constants.ChartRepoIdLabelKey: repo.GetHelmRepoId(),
},
CreationTimestamp: metav1.Time{Time: ver.Created},
CreationTimestamp: metav1.Time{Time: chartVersion.Created},
},
Spec: v1alpha1.HelmApplicationVersionSpec{
Metadata: &v1alpha1.Metadata{
Name: ver.Name,
AppVersion: ver.AppVersion,
Version: ver.Version,
Name: hvw.GetName(),
AppVersion: hvw.GetAppVersion(),
Version: hvw.GetVersion(),
},
URLs: ver.URLs,
Digest: ver.Digest,
URLs: chartVersion.URLs,
Digest: chartVersion.Digest,
Data: chartData,
},
Status: v1alpha1.HelmApplicationVersionStatus{
State: v1alpha1.StateActive,
},
}
c.versions[ver.ApplicationVersionId] = version
}
//modify application category
ctgId := ""
if ctg != "" {
if c.apps[app.ApplicationId].Annotations == nil {
c.apps[app.ApplicationId].Annotations = map[string]string{constants.CategoryIdLabelKey: ctg}
} else {
c.apps[app.ApplicationId].Annotations[constants.CategoryIdLabelKey] = ctg
// It is not necessary to store these pieces of information when this is not a built-in repo.
if helmrepoindex.IsBuiltInRepo(repo.Name) {
version.Spec.Sources = hvw.GetRawSources()
version.Spec.Maintainers = hvw.GetRawMaintainers()
version.Spec.Home = hvw.GetHome()
}
c.versions[chartVersion.ApplicationVersionId] = version
// Find the latest version.
currSemver, err := semver.NewVersion(version.GetSemver())
if err == nil {
if latestSemver == nil {
// the first valid semver
latestSemver = currSemver
latestVersionName = version.GetVersionName()
// Use the category of the latest version as the category of the app.
ctg = chartVersion.Annotations[CategoryAnnotationKey]
} else if latestSemver.LessThan(currSemver) {
// find a newer valid semver
latestSemver = currSemver
latestVersionName = version.GetVersionName()
ctg = chartVersion.Annotations[CategoryAnnotationKey]
}
} else {
// If the semver is invalid, just ignore it.
klog.V(2).Infof("parse version failed, id: %s, err: %s", version.Name, err)
}
ctgId = ctg
} else {
ctgId = v1alpha1.UncategorizedId
}
if _, exists := c.repoCtgCounts[repoId][ctgId]; !exists {
c.repoCtgCounts[repoId][ctgId] = 1
} else {
c.repoCtgCounts[repoId][ctgId] += 1
helmApp.Status.LatestVersion = latestVersionName
if helmrepoindex.IsBuiltInRepo(repo.Name) {
// Add category id to the apps in the built-in repo
ctgId := c.translateCategoryNameToId(ctg)
if helmApp.Labels == nil {
helmApp.Labels = map[string]string{}
}
helmApp.Labels[constants.CategoryIdLabelKey] = ctgId
c.builtinCategoryCounts[ctgId] += 1
}
}
@@ -262,7 +349,7 @@ func (c *cachedRepos) addRepo(repo *v1alpha1.HelmRepo, builtin bool) error {
return nil
}
func (c *cachedRepos) ListApplicationsByRepoId(repoId string) (ret []*v1alpha1.HelmApplication, exists bool) {
func (c *cachedRepos) ListApplicationsInRepo(repoId string) (ret []*v1alpha1.HelmApplication, exists bool) {
c.RLock()
defer c.RUnlock()
@@ -279,6 +366,23 @@ func (c *cachedRepos) ListApplicationsByRepoId(repoId string) (ret []*v1alpha1.H
return ret, true
}
func (c *cachedRepos) ListApplicationsInBuiltinRepo(selector labels.Selector) (ret []*v1alpha1.HelmApplication, exists bool) {
c.RLock()
defer c.RUnlock()
ret = make([]*v1alpha1.HelmApplication, 0, 20)
for _, app := range c.apps {
if strings.HasPrefix(app.GetHelmRepoId(), v1alpha1.BuiltinRepoPrefix) {
if selector != nil && !selector.Empty() &&
(app.Labels == nil || !selector.Matches(labels.Set(app.Labels))) { // If the selector is not empty, we must check whether the labels of the app match the selector.
continue
}
ret = append(ret, app)
}
}
return ret, true
}
func (c *cachedRepos) ListAppVersionsByAppId(appId string) (ret []*v1alpha1.HelmApplicationVersion, exists bool) {
c.RLock()
defer c.RUnlock()

2
staging/src/kubesphere.io/api/application/v1alpha1/constants.go
View File

@@ -49,6 +49,7 @@ const (
HelmApplicationAppStoreSuffix = "-store"
HelmApplicationIdPrefix = "app-"
HelmRepoIdPrefix = "repo-"
BuiltinRepoPrefix = "builtin-"
HelmApplicationVersionIdPrefix = "appv-"
HelmCategoryIdPrefix = "ctg-"
HelmAttachmentPrefix = "att-"
@@ -59,5 +60,6 @@ const (
ApplicationInstance = "app.kubesphere.io/instance"
RepoSyncPeriod = "app.kubesphere.io/sync-period"
OriginWorkspaceLabelKey = "kubesphere.io/workspace-origin"
)