Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: agent get SPD syncly #558

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feat: agent get SPD syncly #558

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0ce43ec
feat: agent get SPD syncly
ddjjia May 4, 2024
b49efba
Merge branch 'main' into feat/spdfetcher-updateSPDCache-from-latest
ddjjia May 15, 2024
d57b8bb
gofumpt
ddjjia May 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions cmd/katalyst-agent/app/options/metaserver/metaserver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ const (
const (
defaultServiceProfileSkipCorruptionError = true
defaultServiceProfileCacheTTL = 1 * time.Minute
defaultGetFromRemote = false
)

const (
Expand Down Expand Up @@ -74,6 +75,7 @@ type MetaServerOptions struct {
// configurations for spd
ServiceProfileSkipCorruptionError bool
ServiceProfileCacheTTL time.Duration
GetFromRemote bool

// configurations for pod-cache
KubeletPodCacheSyncPeriod time.Duration
Expand Down Expand Up @@ -104,6 +106,7 @@ func NewMetaServerOptions() *MetaServerOptions {

ServiceProfileSkipCorruptionError: defaultServiceProfileSkipCorruptionError,
ServiceProfileCacheTTL: defaultServiceProfileCacheTTL,
GetFromRemote: defaultGetFromRemote,

KubeletPodCacheSyncPeriod: defaultKubeletPodCacheSyncPeriod,
KubeletPodCacheSyncMaxRate: defaultKubeletPodCacheSyncMaxRate,
Expand Down Expand Up @@ -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.GetFromRemote, "get-from-remote", o.GetFromRemote, "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")
Expand Down Expand Up @@ -174,6 +178,7 @@ func (o *MetaServerOptions) ApplyTo(c *metaserver.MetaServerConfiguration) error

c.ServiceProfileSkipCorruptionError = o.ServiceProfileSkipCorruptionError
c.ServiceProfileCacheTTL = o.ServiceProfileCacheTTL
c.GetFromRemote = o.GetFromRemote

c.KubeletPodCacheSyncPeriod = o.KubeletPodCacheSyncPeriod
c.KubeletPodCacheSyncMaxRate = rate.Limit(o.KubeletPodCacheSyncMaxRate)
Expand Down
1 change: 1 addition & 0 deletions pkg/config/agent/metaserver/spd.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import "time"
type SPDConfiguration struct {
ServiceProfileSkipCorruptionError bool
ServiceProfileCacheTTL time.Duration
GetFromRemote bool
}

func NewSPDConfiguration() *SPDConfiguration {
Expand Down
38 changes: 31 additions & 7 deletions pkg/metaserver/spd/fetcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,8 +78,9 @@ func (d DummySPDFetcher) Run(_ context.Context) {
}

type spdFetcher struct {
started *atomic.Bool
mux sync.Mutex
started *atomic.Bool
getFromRemote bool
mux sync.Mutex

client *client.GenericClientSet
emitter metrics.MetricEmitter
Expand All @@ -106,6 +107,7 @@ func NewSPDFetcher(clientSet *client.GenericClientSet, emitter metrics.MetricEmi
emitter: emitter,
checkpointManager: checkpointManager,
cncFetcher: cncFetcher,
getFromRemote: conf.GetFromRemote,
}

m.getPodSPDNameFunc = util.GetPodSPDName
Expand Down Expand Up @@ -148,7 +150,7 @@ func (s *spdFetcher) Run(ctx context.Context) {
<-ctx.Done()
}

func (s *spdFetcher) getSPDByNamespaceName(_ context.Context, namespace, name string) (*workloadapis.ServiceProfileDescriptor, error) {
func (s *spdFetcher) getSPDByNamespaceName(ctx context.Context, namespace, name string) (*workloadapis.ServiceProfileDescriptor, error) {
key := native.GenerateNamespaceNameKey(namespace, name)
baseTag := []metrics.MetricTag{
{Key: "spdNamespace", Val: namespace},
Expand All @@ -159,6 +161,28 @@ func (s *spdFetcher) getSPDByNamespaceName(_ context.Context, namespace, name st
currentSPD := s.spdCache.GetSPD(key, true)
if currentSPD != nil {
return currentSPD, nil
} else if s.getFromRemote {
klog.Infof("[spd-manager] need to get spd %s from remote", key)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this logic in the getSPDByNamespaceName can be abstracted into a function with the logic in sync.

targetConfig, err := s.getSPDTargetConfig(ctx, namespace, name)
if err != nil {
klog.Warningf("[spd-manager] get spd targetConfig config failed: %v, use local cache instead", err)
targetConfig = &configapis.TargetConfig{
ConfigNamespace: namespace,
ConfigName: name,
}
_ = s.emitter.StoreInt64(metricsNameGetCNCTargetConfigFailed, 1, metrics.MetricTypeNameCount, baseTag...)
}
err = s.updateSPDCacheIfNeed(ctx, currentSPD, targetConfig, s.getFromRemote)
if err != nil {
klog.Errorf("[spd-manager] failed update spd cache from remote: %v, use local cache instead", err)
_ = s.emitter.StoreInt64(metricsNameUpdateCacheFailed, 1, metrics.MetricTypeNameCount, baseTag...)
return currentSPD, nil
}

currentSPD = s.spdCache.GetSPD(key, true)
if currentSPD != nil {
return currentSPD, nil
}
}

_ = s.emitter.StoreInt64(metricsNameCacheNotFound, 1, metrics.MetricTypeNameCount, baseTag...)
Expand Down Expand Up @@ -212,7 +236,7 @@ func (s *spdFetcher) sync(ctx context.Context) {

// try to update spd cache from remote if cache spd hash is not equal to target config hash,
// the rate of getting remote spd will be limited by spd ServiceProfileCacheTTL
err = s.updateSPDCacheIfNeed(ctx, originSPD, targetConfig)
err = s.updateSPDCacheIfNeed(ctx, originSPD, targetConfig, false)
if err != nil {
klog.Errorf("[spd-manager] failed update spd cache from remote: %v, use local cache instead", err)
_ = s.emitter.StoreInt64(metricsNameUpdateCacheFailed, 1, metrics.MetricTypeNameCount, baseTag...)
Expand All @@ -223,18 +247,18 @@ func (s *spdFetcher) sync(ctx context.Context) {
// updateSPDCacheIfNeed checks if the previous spd has changed, and
// re-get from APIServer if the previous is out-of date.
func (s *spdFetcher) updateSPDCacheIfNeed(ctx context.Context, originSPD *workloadapis.ServiceProfileDescriptor,
targetConfig *configapis.TargetConfig,
targetConfig *configapis.TargetConfig, needToGetFromRemote bool,
) error {
if originSPD == nil && targetConfig == nil {
return nil
}

now := time.Now()
if originSPD == nil || util.GetSPDHash(originSPD) != targetConfig.Hash {
if originSPD == nil || util.GetSPDHash(originSPD) != targetConfig.Hash || needToGetFromRemote {
key := native.GenerateNamespaceNameKey(targetConfig.ConfigNamespace, targetConfig.ConfigName)
// Skip the backoff delay if the configuration hash of the CNC target changes, ensuring the
// local SPD cache is always updated with the latest configuration.
if nextFetchRemoteTime := s.spdCache.GetNextFetchRemoteTime(key, now, targetConfig.Hash != ""); nextFetchRemoteTime.After(time.Now()) {
if nextFetchRemoteTime := s.spdCache.GetNextFetchRemoteTime(key, now, targetConfig.Hash != ""); nextFetchRemoteTime.After(time.Now()) && !needToGetFromRemote {
return nil
} else {
// first update the timestamp of the last attempt to fetch the remote spd to
Expand Down
1 change: 1 addition & 0 deletions pkg/metaserver/spd/fetcher_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ func generateTestConfiguration(t *testing.T, nodeName string, checkpoint string)

testConfiguration.NodeName = nodeName
testConfiguration.CheckpointManagerDir = checkpoint
testConfiguration.GetFromRemote = true
return testConfiguration
}

Expand Down
Loading