add checkpoint for npd fetcher

Signed-off-by: xudong.l <xudong.l@bytedance.com>

Author: XudongLiuHarold

pkg/metaserver/metaserver.go

@@ -81,23 +81,20 @@ func NewMetaServer(clientSet *client.GenericClientSet, emitter metrics.MetricEmi
 
 	var npdFetcher npd.NPDFetcher
 	if conf.EnableNPDFetcher {
-		npdFetcher = npd.NewNPDFetcher(clientSet, metaAgent.CNCFetcher, conf.KCCConfiguration)
+		npdFetcher, err = npd.NewNPDFetcher(clientSet, metaAgent.CNCFetcher, conf, emitter)
+		if err != nil {
+			return nil, fmt.Errorf("initializes npd fetcher failed: %s", err)
+		}
 	} else {
 		npdFetcher = npd.NewDummyNPDFetcher()
 	}
 
-	var resourcePackageManager resourcepackage.ResourcePackageManager
-	resourcePackageManager, err = resourcepackage.NewResourcePackageManager(npdFetcher, conf, emitter)
-	if err != nil {
-		return nil, fmt.Errorf("initializes resource package manager failed: %s", err)
-	}
-
 	return &MetaServer{
 		MetaAgent:               metaAgent,
 		ConfigurationManager:    configurationManager,
 		ServiceProfilingManager: spd.NewServiceProfilingManager(spdFetcher),
 		ExternalManager:         external.InitExternalManager(metaAgent.PodFetcher),
-		ResourcePackageManager:  resourcePackageManager,
+		ResourcePackageManager:  resourcepackage.NewResourcePackageManager(npdFetcher),
 		NPDFetcher:              npdFetcher,
 	}, nil
 }
@@ -114,7 +111,6 @@ func (m *MetaServer) Run(ctx context.Context) {
 	go m.ConfigurationManager.Run(ctx)
 	go m.ServiceProfilingManager.Run(ctx)
 	go m.ExternalManager.Run(ctx)
-	go m.ResourcePackageManager.Run(ctx)
 
 	m.Unlock()
 	<-ctx.Done()

pkg/metaserver/metaserver_test.go

@@ -37,6 +37,7 @@ import (
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/pod"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/external"
 	dynamicconfig "github.com/kubewharf/katalyst-core/pkg/metaserver/kcc"
+	"github.com/kubewharf/katalyst-core/pkg/metaserver/npd"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/resourcepackage"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/spd"
 	"github.com/kubewharf/katalyst-core/pkg/metrics"
@@ -61,10 +62,11 @@ func generateTestMetaServer(clientSet *client.GenericClientSet, conf *config.Con
 				metrics.DummyMetrics{}, &pod.PodFetcherStub{}, &machine.KatalystMachineInfo{}),
 			AgentConf: conf.MetaServerConfiguration.AgentConfiguration,
 		},
+		NPDFetcher:              npd.NewDummyNPDFetcher(),
 		ConfigurationManager:    &dynamicconfig.DummyConfigurationManager{},
 		ServiceProfilingManager: &spd.DummyServiceProfilingManager{},
 		ExternalManager:         &external.DummyExternalManager{},
-		ResourcePackageManager:  &resourcepackage.DummyResourcePackageManager{},
+		ResourcePackageManager:  resourcepackage.NewResourcePackageManager(npd.NewDummyNPDFetcher()),
 	}
 }
 

…metaserver/resourcepackage/checkpoint.go pkg/metaserver/npd/checkpoint.gopkg/metaserver/resourcepackage/checkpoint.go renamed to pkg/metaserver/npd/checkpoint.go pkg/metaserver/resourcepackage/checkpoint.go renamed to pkg/metaserver/npd/checkpoint.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package resourcepackage
+package npd
 
 import (
 	"encoding/json"

…erver/resourcepackage/checkpoint_test.go pkg/metaserver/npd/checkpoint_test.gopkg/metaserver/resourcepackage/checkpoint_test.go renamed to pkg/metaserver/npd/checkpoint_test.go pkg/metaserver/resourcepackage/checkpoint_test.go renamed to pkg/metaserver/npd/checkpoint_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package resourcepackage
+package npd
 
 import (
 	"testing"

pkg/metaserver/npd/fetcher.go

@@ -18,15 +18,36 @@ package npd
 
 import (
 	"context"
+	"fmt"
+	"time"
+
+	"github.com/pkg/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager"
 
 	"github.com/kubewharf/katalyst-api/pkg/apis/node/v1alpha1"
 	"github.com/kubewharf/katalyst-core/pkg/client"
-	metaconfig "github.com/kubewharf/katalyst-core/pkg/config/agent/metaserver"
+	pkgconfig "github.com/kubewharf/katalyst-core/pkg/config"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/agent/cnc"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/kcc"
+	"github.com/kubewharf/katalyst-core/pkg/metrics"
 	"github.com/kubewharf/katalyst-core/pkg/util"
 )
 
+const (
+	metricsNameUpdateNPD         = "metaserver_update_npd"
+	metricsNameLoadNPDCheckpoint = "metaserver_load_npd_checkpoint"
+
+	metricsValueStatusCheckpointNotFoundOrCorrupted = "notFoundOrCorrupted"
+	metricsValueStatusCheckpointInvalidOrExpired    = "invalidOrExpired"
+	metricsValueStatusCheckpointSuccess             = "success"
+)
+
+const (
+	npdFetcherCheckpoint = "npd_fetcher_checkpoint"
+)
+
 type NPDFetcher interface {
 	GetNPD(ctx context.Context) (*v1alpha1.NodeProfileDescriptor, error)
 }
@@ -41,26 +62,110 @@ func (f *DummyNPDFetcher) GetNPD(_ context.Context) (*v1alpha1.NodeProfileDescri
 
 type npdFetcher struct {
 	configLoader kcc.ConfigurationLoader
+
+	// checkpoint stores recent fetched NPDs
+	checkpointManager checkpointmanager.CheckpointManager
+
+	// checkpointGraceTime is the duration to consider a checkpoint valid
+	checkpointGraceTime time.Duration
+
+	// emitter is the metric emitter
+	emitter metrics.MetricEmitter
 }
 
 func NewNPDFetcher(clientSet *client.GenericClientSet,
 	cncFetcher cnc.CNCFetcher,
-	conf *metaconfig.KCCConfiguration,
-) NPDFetcher {
+	conf *pkgconfig.Configuration,
+	emitter metrics.MetricEmitter,
+) (NPDFetcher, error) {
+	checkpointManager, err := checkpointmanager.NewCheckpointManager(conf.CheckpointManagerDir)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize checkpoint manager: %v", err)
+	}
+
 	configLoader := kcc.NewKatalystCustomConfigLoader(clientSet, conf.ConfigCacheTTL, cncFetcher)
 	return &npdFetcher{
-		configLoader: configLoader,
-	}
+		configLoader:        configLoader,
+		checkpointManager:   checkpointManager,
+		checkpointGraceTime: conf.ConfigCheckpointGraceTime,
+		emitter:             emitter,
+	}, nil
 }
 
 func NewDummyNPDFetcher() *DummyNPDFetcher {
 	return &DummyNPDFetcher{}
 }
 
 func (f *npdFetcher) GetNPD(ctx context.Context) (*v1alpha1.NodeProfileDescriptor, error) {
+	latestNpd, err := f.getNPD(ctx)
+	if err != nil || latestNpd == nil {
+		klog.Errorf("[npd-fetcher] try get new npd error: %v", err)
+		// load from checkpoint if fetcher fails
+		data, err := f.readCheckpoint()
+		if err != nil {
+			_ = f.emitter.StoreInt64(metricsNameUpdateNPD, 1, metrics.MetricTypeNameRaw, []metrics.MetricTag{
+				{Key: "status", Val: metricsValueStatusCheckpointNotFoundOrCorrupted},
+			}...)
+			return nil, errors.Wrap(err, "failed to load npd checkpoint")
+		}
+		npdData, timestamp := data.GetProfile()
+		if time.Now().Before(timestamp.Add(f.checkpointGraceTime)) && npdData != nil {
+			latestNpd = npdData
+			klog.Infof("[npd-fetcher] failed to load npd from remote, use local checkpoint instead")
+			_ = f.emitter.StoreInt64(metricsNameLoadNPDCheckpoint, 1, metrics.MetricTypeNameRaw, []metrics.MetricTag{
+				{Key: "status", Val: metricsValueStatusCheckpointSuccess},
+				{Key: "npd", Val: latestNpd.Name},
+			}...)
+		} else {
+			_ = f.emitter.StoreInt64(metricsNameLoadNPDCheckpoint, 1, metrics.MetricTypeNameRaw, []metrics.MetricTag{
+				{Key: "status", Val: metricsValueStatusCheckpointInvalidOrExpired},
+				{Key: "npd", Val: "invalid"},
+			}...)
+			return nil, fmt.Errorf("failed to load npd checkpoint, checkpoint expired or invalid")
+		}
+	}
+	f.writeCheckpoint(latestNpd)
+	return latestNpd, nil
+}
+
+// getNPD retrieves the latest NodeProfileDescriptor from the config loader.
+func (f *npdFetcher) getNPD(ctx context.Context) (*v1alpha1.NodeProfileDescriptor, error) {
 	npd := &v1alpha1.NodeProfileDescriptor{}
 	if err := f.configLoader.LoadConfig(ctx, util.NPDGVR, npd); err != nil {
 		return nil, err
 	}
 	return npd, nil
 }
+
+// readCheckpoint reads the checkpoint from disk
+func (f *npdFetcher) readCheckpoint() (NodeProfileCheckpoint, error) {
+	cp := NewCheckpoint(NodeProfileData{})
+	if err := f.checkpointManager.GetCheckpoint(npdFetcherCheckpoint, cp); err != nil {
+		klog.Errorf("[npd-fetcher] failed to get npd fetcher checkpoint: %v", err)
+		return nil, err
+	}
+	return cp, nil
+}
+
+// writeCheckpoint writes the checkpoint to disk
+func (f *npdFetcher) writeCheckpoint(npd *v1alpha1.NodeProfileDescriptor) {
+	data, err := f.readCheckpoint()
+	if err != nil {
+		klog.Errorf("[npd-fetcher] load checkpoint from %q failed: %v, try to overwrite it", npdFetcherCheckpoint, err)
+		_ = f.emitter.StoreInt64(metricsNameLoadNPDCheckpoint, 1, metrics.MetricTypeNameRaw, []metrics.MetricTag{
+			{Key: "status", Val: metricsValueStatusCheckpointNotFoundOrCorrupted},
+			{Key: "node", Val: npd.Name},
+		}...)
+	}
+	// checkpoint doesn't exist or became corrupted, make a new checkpoint
+	if data == nil {
+		data = NewCheckpoint(NodeProfileData{})
+	}
+
+	// set config value and timestamp for kind
+	data.SetProfile(npd, metav1.Now())
+	err = f.checkpointManager.CreateCheckpoint(npdFetcherCheckpoint, data)
+	if err != nil {
+		klog.Errorf("[npd-fetcher] failed to write checkpoint file %q: %v", npdFetcherCheckpoint, err)
+	}
+}