agent get SPD syncly & spd creation improvement

cmd/katalyst-agent/app/options/metaserver/metaserver.go

@@ -41,6 +41,7 @@ const (
 const (
 	defaultServiceProfileSkipCorruptionError = true
 	defaultServiceProfileCacheTTL            = 1 * time.Minute
+	defaultSPDGetFromRemote                  = false
 )
 
 const (
@@ -74,6 +75,7 @@ type MetaServerOptions struct {
 	// configurations for spd
 	ServiceProfileSkipCorruptionError bool
 	ServiceProfileCacheTTL            time.Duration
+	SPDGetFromRemote                  bool
 
 	// configurations for pod-cache
 	KubeletPodCacheSyncPeriod    time.Duration
@@ -104,6 +106,7 @@ func NewMetaServerOptions() *MetaServerOptions {
 
 		ServiceProfileSkipCorruptionError: defaultServiceProfileSkipCorruptionError,
 		ServiceProfileCacheTTL:            defaultServiceProfileCacheTTL,
+		SPDGetFromRemote:                  defaultSPDGetFromRemote,
 
 		KubeletPodCacheSyncPeriod:    defaultKubeletPodCacheSyncPeriod,
 		KubeletPodCacheSyncMaxRate:   defaultKubeletPodCacheSyncMaxRate,
@@ -142,6 +145,7 @@ func (o *MetaServerOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 		"Whether to skip corruption error when loading spd checkpoint")
 	fs.DurationVar(&o.ServiceProfileCacheTTL, "service-profile-cache-ttl", o.ServiceProfileCacheTTL,
 		"The ttl of service profile manager cache remote spd")
+	fs.BoolVar(&o.SPDGetFromRemote, "spd-get-from-remote", o.SPDGetFromRemote, "get spd from remote if not in cache")
 
 	fs.DurationVar(&o.KubeletPodCacheSyncPeriod, "kubelet-pod-cache-sync-period", o.KubeletPodCacheSyncPeriod,
 		"The period of meta server to sync pod from kubelet 10255 port")
@@ -174,6 +178,7 @@ func (o *MetaServerOptions) ApplyTo(c *metaserver.MetaServerConfiguration) error
 
 	c.ServiceProfileSkipCorruptionError = o.ServiceProfileSkipCorruptionError
 	c.ServiceProfileCacheTTL = o.ServiceProfileCacheTTL
+	c.SPDGetFromRemote = o.SPDGetFromRemote
 
 	c.KubeletPodCacheSyncPeriod = o.KubeletPodCacheSyncPeriod
 	c.KubeletPodCacheSyncMaxRate = rate.Limit(o.KubeletPodCacheSyncMaxRate)

pkg/config/agent/metaserver/spd.go

@@ -21,6 +21,7 @@ import "time"
 type SPDConfiguration struct {
 	ServiceProfileSkipCorruptionError bool
 	ServiceProfileCacheTTL            time.Duration
+	SPDGetFromRemote                  bool
 }
 
 func NewSPDConfiguration() *SPDConfiguration {

pkg/controller/spd/indicator-plugin/manager.go

@@ -34,7 +34,7 @@ const (
 // IndicatorUpdater is used by IndicatorPlugin as a unified implementation
 // to trigger indicator updating logic.
 type IndicatorUpdater interface {
-	// UpdateExtendedIndicatorSpec + UpdateBusinessIndicatorSpec + UpdateSystemIndicatorSpec + UpdateBusinessIndicatorStatus
+	// UpdateExtendedIndicatorSpec + UpdateBusinessIndicatorSpec + UpdateSystemIndicatorSpec + UpdateBusinessIndicatorStatus + UpdateAggMetrics
 	// for indicator add functions, IndicatorUpdater will try to merge them in local stores.
 	UpdateExtendedIndicatorSpec(_ types.NamespacedName, _ []apiworkload.ServiceExtendedIndicatorSpec)
 	UpdateBusinessIndicatorSpec(_ types.NamespacedName, _ []apiworkload.ServiceBusinessIndicatorSpec)
@@ -136,15 +136,15 @@ func (u *IndicatorManager) UpdateSystemIndicatorSpec(nn types.NamespacedName, in
 	}
 }
 
-func (u *IndicatorManager) UpdateBusinessIndicatorStatus(nn types.NamespacedName, indicators []apiworkload.ServiceBusinessIndicatorStatus) {
+func (u *IndicatorManager) UpdateBusinessIndicatorStatus(nn types.NamespacedName, businessIndicators []apiworkload.ServiceBusinessIndicatorStatus) {
 	u.statusMtx.Lock()
 
 	insert := false
 	if _, ok := u.statusMap[nn]; !ok {
 		insert = true
 		u.statusMap[nn] = initServiceProfileDescriptorStatus()
 	}
-	for _, indicator := range indicators {
+	for _, indicator := range businessIndicators {
 		util.InsertSPDBusinessIndicatorStatus(u.statusMap[nn], &indicator)
 	}
 
@@ -223,5 +223,6 @@ func initServiceProfileDescriptorSpec() *apiworkload.ServiceProfileDescriptorSpe
 func initServiceProfileDescriptorStatus() *apiworkload.ServiceProfileDescriptorStatus {
 	return &apiworkload.ServiceProfileDescriptorStatus{
 		BusinessStatus: []apiworkload.ServiceBusinessIndicatorStatus{},
+		AggMetrics:     []apiworkload.AggPodMetrics{},
 	}
 }

pkg/controller/spd/indicator-plugin/plugin.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"sync"
 
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	apiworkload "github.com/kubewharf/katalyst-api/pkg/apis/workload/v1alpha1"
@@ -44,6 +45,9 @@ type IndicatorPlugin interface {
 	GetSupportedExtendedIndicatorSpec() []string
 	GetSupportedBusinessIndicatorStatus() []apiworkload.ServiceBusinessIndicatorName
 	GetSupportedAggMetricsStatus() []string
+	// GetAggMetrics is only used in SPD creation for timing-sensitive status.
+	// The other non-timing-sensitive status are updated at UpdateAggMetrics in indicatorUpdater.
+	GetAggMetrics(workload *unstructured.Unstructured) ([]apiworkload.AggPodMetrics, error)
 }
 
 type DummyIndicatorPlugin struct {
@@ -78,6 +82,10 @@ func (d DummyIndicatorPlugin) GetSupportedAggMetricsStatus() []string {
 	return nil
 }
 
+func (d DummyIndicatorPlugin) GetAggMetrics(_ *unstructured.Unstructured) ([]apiworkload.AggPodMetrics, error) {
+	return nil, nil
+}
+
 // pluginInitializers is used to store the initializing function for each plugin
 var pluginInitializers sync.Map
 

pkg/controller/spd/indicator-plugin/plugins/resource-portrait/indicator_plugin.go

@@ -121,6 +121,10 @@ func (p *ResourcePortraitIndicatorPlugin) GetSupportedAggMetricsStatus() []strin
 	return []string{resourcePortraitExtendedSpecName}
 }
 
+func (p *ResourcePortraitIndicatorPlugin) GetAggMetrics(_ *unstructured.Unstructured) ([]apiworkload.AggPodMetrics, error) {
+	return nil, nil
+}
+
 // resyncSpecWorker is used to synchronize global configuration to SPD.
 func (p *ResourcePortraitIndicatorPlugin) resyncSpecWorker() {
 	spdList, err := p.spdLister.List(labels.Everything())