原生yaml资源管理 (#801)

* 原生yaml资源管理

* fix

* fix

---------

Co-authored-by: hannatao <413024870@qq.com>

ami/ami.go

@@ -13,6 +13,7 @@ import (
 	"github.com/baetyl/baetyl-go/v2/log"
 	specv1 "github.com/baetyl/baetyl-go/v2/spec/v1"
 	"github.com/gorilla/websocket"
+	bh "github.com/timshannon/bolthold"
 
 	"github.com/baetyl/baetyl/v2/config"
 	"github.com/baetyl/baetyl/v2/utils"
@@ -33,7 +34,7 @@ var Hooks = map[string]interface{}{}
 
 type CollectStatsExtFunc func(mode string) (map[string]interface{}, error)
 
-type New func(cfg config.AmiConfig) (AMI, error)
+type New func(cfg config.AmiConfig, sto *bh.Store) (AMI, error)
 
 type Pipe struct {
 	InReader  *io.PipeReader
@@ -53,7 +54,7 @@ type AMI interface {
 
 	// app
 	ApplyApp(string, specv1.Application, map[string]specv1.Configuration, map[string]specv1.Secret) error
-	DeleteApp(string, string) error
+	DeleteApp(string, specv1.AppInfo) error
 	StatsApps(string) ([]specv1.AppStats, error)
 
 	FetchLog(namespace, pod, container string, tailLines, sinceSeconds int64) ([]byte, error)
@@ -119,7 +120,7 @@ type ProcessInfo struct {
 	Ppid int32
 }
 
-func NewAMI(mode string, cfg config.AmiConfig) (AMI, error) {
+func NewAMI(mode string, cfg config.AmiConfig, sto *bh.Store) (AMI, error) {
 	mu.Lock()
 	defer mu.Unlock()
 	if ami, ok := amiImpls[mode]; ok {
@@ -129,7 +130,7 @@ func NewAMI(mode string, cfg config.AmiConfig) (AMI, error) {
 	if !ok {
 		return nil, errors.Trace(os.ErrInvalid)
 	}
-	ami, err := amiNew(cfg)
+	ami, err := amiNew(cfg, sto)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

ami/kube/kube.go

@@ -30,8 +30,8 @@ func init() {
 	ami.Register("kubernetes", newKubeImpl)
 }
 
-func newKubeImpl(cfg config.AmiConfig) (ami.AMI, error) {
-	kubeCli, err := newClient(cfg.Kube)
+func newKubeImpl(cfg config.AmiConfig, sto *bh.Store) (ami.AMI, error) {
+	cli, err := newClient(cfg.Kube)
 	if err != nil {
 		return nil, err
 	}
@@ -44,11 +44,12 @@ func newKubeImpl(cfg config.AmiConfig) (ami.AMI, error) {
 	}
 
 	model := &kubeImpl{
-		knn:  knn,
-		cli:  kubeCli,
-		helm: actionConfig,
-		conf: &cfg.Kube,
-		log:  logv2.With(logv2.Any("ami", "kube")),
+		knn:   knn,
+		cli:   cli,
+		helm:  actionConfig,
+		store: sto,
+		conf:  &cfg.Kube,
+		log:   logv2.With(logv2.Any("ami", "kube")),
 	}
 	return model, nil
 }
@@ -57,11 +58,18 @@ func (k *kubeImpl) ApplyApp(ns string, app specv1.Application, cfgs map[string]s
 	if app.Type == specv1.AppTypeHelm {
 		return k.ApplyHelm(ns, app, cfgs)
 	}
+	if app.Type == specv1.AppTypeYaml {
+		ns = app.Labels[specv1.CustomAppNsLabel]
+	}
 	err := k.checkAndCreateNamespace(ns)
 	if err != nil {
 		return errors.Trace(err)
 	}
-	if err = k.applyConfigurations(ns, cfgs); err != nil {
+
+	if app.Type == specv1.AppTypeYaml {
+		return k.ApplyYaml(app, cfgs)
+	}
+	if err := k.applyConfigurations(ns, cfgs); err != nil {
 		return errors.Trace(err)
 	}
 	if err = k.applySecrets(ns, secs); err != nil {
@@ -83,15 +91,32 @@ func (k *kubeImpl) ApplyApp(ns string, app specv1.Application, cfgs map[string]s
 	return nil
 }
 
-func (k *kubeImpl) DeleteApp(ns string, app string) error {
+func makeKey(kind specv1.Kind, name, ver string) string {
+	if name == "" || ver == "" {
+		return ""
+	}
+	return string(kind) + "-" + name + "-" + ver
+}
+
+func (k *kubeImpl) DeleteApp(ns string, app specv1.AppInfo) error {
 	if ns == context.EdgeNamespace() {
-		err := k.DeleteHelm(ns, app)
+		err := k.DeleteHelm(ns, app.Name)
 		// If delete helm success or err is not ErrNotHelmApp, return directly
 		if err == nil || err.Error() != ErrNotHelmApp {
 			return err
 		}
 	}
-	return k.deleteApplication(ns, app)
+	delApp := new(specv1.Application)
+	key := makeKey(specv1.KindApplication, app.Name, app.Version)
+	err := k.store.Get(key, delApp)
+	if err != nil {
+		return err
+	}
+	// delete yaml app
+	if delApp.Type == specv1.AppTypeYaml {
+		return k.DeleteYaml(delApp)
+	}
+	return k.deleteApplication(ns, app.Name)
 }
 
 func (k *kubeImpl) StatsApps(ns string) ([]specv1.AppStats, error) {
@@ -125,13 +150,19 @@ func (k *kubeImpl) StatsApps(ns string) ([]specv1.AppStats, error) {
 		return nil, errors.Trace(err)
 	}
 	res = append(res, js...)
-	// Collect only in baetyl-edge namespace
+
 	if ns == context.EdgeNamespace() {
 		helmStats, err := k.StatsHelm(ns)
 		if err != nil {
-			return nil, errors.Trace(err)
+			return res, errors.Trace(err)
 		}
 		res = append(res, helmStats...)
+
+		yamlStats, err := k.StatsYaml()
+		if err != nil {
+			return res, errors.Trace(err)
+		}
+		res = append(res, yamlStats...)
 	}
 	return res, nil
 }

ami/kube/kube_client.go

@@ -3,6 +3,7 @@ package kube
 import (
 	"github.com/baetyl/baetyl-go/v2/errors"
 	"k8s.io/client-go/discovery"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	appv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
 	v2 "k8s.io/client-go/kubernetes/typed/autoscaling/v2"
@@ -24,6 +25,7 @@ type client struct {
 	metrics    metricsv1beta1.MetricsV1beta1Interface
 	discovery  discovery.DiscoveryInterface
 	autoscale  v2.AutoscalingV2Interface
+	dynamic    *dynamic.DynamicClient
 }
 
 func newClient(cfg config.KubeConfig) (*client, error) {
@@ -45,6 +47,11 @@ func newClient(cfg config.KubeConfig) (*client, error) {
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+
+	dynamicClient, err := dynamic.NewForConfig(kubeConfig)
+	if err != nil {
+		panic(err.Error())
+	}
 	return &client{
 		kubeConfig: kubeConfig,
 		core:       kubeClient.CoreV1(),
@@ -53,5 +60,6 @@ func newClient(cfg config.KubeConfig) (*client, error) {
 		metrics:    metricsCli.MetricsV1beta1(),
 		discovery:  kubeClient.Discovery(),
 		autoscale:  kubeClient.AutoscalingV2(),
+		dynamic:    dynamicClient,
 	}, nil
 }

ami/kube/kube_test.go

@@ -12,6 +12,6 @@ func Test_newKubeImpl(t *testing.T) {
 	c := config.AmiConfig{}
 	c.Kube.OutCluster = false
 
-	_, err := newKubeImpl(c)
+	_, err := newKubeImpl(c, nil)
 	assert.Error(t, err)
 }