reconcile npd for resource package manager

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

Author: XudongLiuHarold

pkg/metaserver/metaserver.go

@@ -86,12 +86,18 @@ func NewMetaServer(clientSet *client.GenericClientSet, emitter metrics.MetricEmi
 		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:  resourcepackage.NewResourcePackageManager(npdFetcher),
+		ResourcePackageManager:  resourcePackageManager,
 		NPDFetcher:              npdFetcher,
 	}, nil
 }
@@ -108,6 +114,7 @@ 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/resourcepackage"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver/spd"
 	"github.com/kubewharf/katalyst-core/pkg/metrics"
 	"github.com/kubewharf/katalyst-core/pkg/util/machine"
@@ -63,6 +64,7 @@ func generateTestMetaServer(clientSet *client.GenericClientSet, conf *config.Con
 		ConfigurationManager:    &dynamicconfig.DummyConfigurationManager{},
 		ServiceProfilingManager: &spd.DummyServiceProfilingManager{},
 		ExternalManager:         &external.DummyExternalManager{},
+		ResourcePackageManager:  &resourcepackage.DummyResourcePackageManager{},
 	}
 }
 

pkg/metaserver/resourcepackage/checkpoint.go

@@ -0,0 +1,101 @@
+/*
+Copyright 2022 The Katalyst Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resourcepackage
+
+import (
+	"encoding/json"
+	"sync"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager"
+	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager/checksum"
+
+	nodev1alpha1 "github.com/kubewharf/katalyst-api/pkg/apis/node/v1alpha1"
+)
+
+type NodeProfileCheckpoint interface {
+	checkpointmanager.Checkpoint
+	GetProfile() (*nodev1alpha1.NodeProfileDescriptor, metav1.Time)
+	SetProfile(npd *nodev1alpha1.NodeProfileDescriptor, t metav1.Time)
+}
+
+type NodeProfileData struct {
+	// Value only store spec of dynamic config crd
+	Value     *nodev1alpha1.NodeProfileDescriptor
+	Timestamp int64
+}
+
+// Data holds checkpoint data and its checksum
+type Data struct {
+	sync.Mutex
+	Item *DataItem
+}
+
+// DataItem stores the checkpoint data and checksum.
+type DataItem struct {
+	Data     NodeProfileData   `json:"data"`
+	Checksum checksum.Checksum `json:"checksum"`
+}
+
+// NewCheckpoint returns an instance of Checkpoint
+func NewCheckpoint(npdData NodeProfileData) NodeProfileCheckpoint {
+	return &Data{
+		Item: &DataItem{
+			Data: npdData,
+		},
+	}
+}
+
+func (d *Data) MarshalCheckpoint() ([]byte, error) {
+	d.Lock()
+	defer d.Unlock()
+
+	d.Item.Checksum = checksum.New(d.Item.Data)
+	return json.Marshal(*(d.Item))
+}
+
+func (d *Data) UnmarshalCheckpoint(blob []byte) error {
+	d.Lock()
+	defer d.Unlock()
+
+	return json.Unmarshal(blob, d.Item)
+}
+
+func (d *Data) VerifyChecksum() error {
+	d.Lock()
+	defer d.Unlock()
+
+	return d.Item.Checksum.Verify(d.Item.Data)
+}
+
+// GetProfile retrieves a NodeProfileDescriptor.
+func (d *Data) GetProfile() (*nodev1alpha1.NodeProfileDescriptor, metav1.Time) {
+	d.Lock()
+	defer d.Unlock()
+	return d.Item.Data.Value, metav1.Unix(d.Item.Data.Timestamp, 0)
+}
+
+// SetProfile sets a NodeProfileDescriptor for a node.
+func (d *Data) SetProfile(npd *nodev1alpha1.NodeProfileDescriptor, t metav1.Time) {
+	d.Lock()
+	defer d.Unlock()
+
+	d.Item.Data = NodeProfileData{
+		Value:     npd,
+		Timestamp: t.Unix(),
+	}
+}

pkg/metaserver/resourcepackage/checkpoint_test.go

@@ -0,0 +1,71 @@
+/*
+Copyright 2022 The Katalyst Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resourcepackage
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	nodev1alpha1 "github.com/kubewharf/katalyst-api/pkg/apis/node/v1alpha1"
+)
+
+func TestNewCheckpoint(t *testing.T) {
+	t.Parallel()
+
+	now := metav1.Now()
+
+	// Prepare a test NodeProfileDescriptor
+	npd := &nodev1alpha1.NodeProfileDescriptor{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "node-a",
+			Namespace: "default",
+		},
+		Spec: nodev1alpha1.NodeProfileDescriptorSpec{},
+		Status: nodev1alpha1.NodeProfileDescriptorStatus{
+			NodeMetrics: []nodev1alpha1.ScopedNodeMetrics{
+				{
+					Scope: "resource-package",
+				},
+			},
+		},
+	}
+
+	// Create a new checkpoint and set data
+	cp := NewCheckpoint(NodeProfileData{})
+	cp.SetProfile(npd, now)
+
+	// Marshal the checkpoint
+	checkpointBytes, err := cp.MarshalCheckpoint()
+	assert.NoError(t, err, "should marshal checkpoint successfully")
+
+	// Unmarshal into a new checkpoint object
+	loaded := &Data{Item: &DataItem{}}
+	err = loaded.UnmarshalCheckpoint(checkpointBytes)
+	assert.NoError(t, err, "should unmarshal checkpoint successfully")
+
+	// Verify checksum integrity
+	err = loaded.VerifyChecksum()
+	assert.NoError(t, err, "checksum verification should pass")
+
+	// Retrieve stored profile and timestamp
+	restoredProfile, ts := loaded.GetProfile()
+
+	assert.Equal(t, metav1.Unix(now.Unix(), 0), ts, "timestamp should match")
+	assert.Equal(t, npd, restoredProfile, "restored profile should match original")
+}