This is an automated email from the ASF dual-hosted git repository.
wilfred-s pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new 5dae0cbc [YUNIKORN-3347] Improve checksum generation and checks (#1126)
5dae0cbc is described below
commit 5dae0cbc70a38dddb827906b6e8fd6a296ce69eb
Author: sidbroski <[email protected]>
AuthorDate: Thu Aug 20 19:24:29 2026 +1000
[YUNIKORN-3347] Improve checksum generation and checks (#1126)
Improved checksum generation and checks, implemented efficient detection
of the config checksum at the start or end of the serialised config,
override missing/incorrect checksum in memory with logging,
and return the checksum plus comparison result from validateConf.
Closes: #1126
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/common/configs/config.go | 77 ++++++++++++++++++++++++++++++++-------
pkg/common/configs/config_test.go | 50 +++++++++++++++++++++++--
pkg/scheduler/context.go | 7 +++-
pkg/webservice/dao/config_info.go | 6 ++-
pkg/webservice/handlers.go | 16 ++++++--
pkg/webservice/handlers_test.go | 33 +++++++++++++++++
6 files changed, 165 insertions(+), 24 deletions(-)
diff --git a/pkg/common/configs/config.go b/pkg/common/configs/config.go
index 50a35fe6..ad0097a8 100644
--- a/pkg/common/configs/config.go
+++ b/pkg/common/configs/config.go
@@ -24,7 +24,6 @@ import (
"errors"
"fmt"
"io"
- "strings"
"go.uber.org/zap"
"go.yaml.in/yaml/v3"
@@ -32,6 +31,19 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
)
+const (
+ // checksumKey is the YAML key that holds the checksum in a serialised
scheduler config.
+ checksumKey = "checksum:"
+ // partitionsKey is the first YAML key of a serialised scheduler
config. It must always be present
+ // and is used to locate the start of the config when the checksum is
stored before it.
+ partitionsKey = "partitions:"
+ // checksumScanWindow is the number of bytes scanned at the start and
the end of a serialised config
+ // to locate the checksum line. Any checksum line would never exceed
this size ("checksum: " plus a 64 character
+ // hex string with an optional trailing newline and quotes).
+ // Scanning only the head and the tail keeps the detection cheap on
large configs which could be megabytes in size.
+ checksumScanWindow = 80
+)
+
// SchedulerConfig can contain multiple partitions. Each partition contains
the queue definition for a logical
// set of scheduler resources.
type SchedulerConfig struct {
@@ -168,9 +180,25 @@ func LoadSchedulerConfigFromByteArray(content []byte)
(*SchedulerConfig, error)
return conf, err
}
+// SetChecksum calculates the sha256 checksum for the serialised config and
stores it in the config.
+// The config might already contain a checksum read from the YAML, it could be
missing or set to an incorrect value.
+// The correct checksum will always override it, the action taken is logged so
that an incorrect checksum can be traced.
func SetChecksum(content []byte, conf *SchedulerConfig) {
- noChecksumContent := GetConfigurationString(content)
- conf.Checksum = fmt.Sprintf("%X",
sha256.Sum256([]byte(noChecksumContent)))
+ // nil safety
+ if conf == nil {
+ return
+ }
+ checksum := fmt.Sprintf("%X",
sha256.Sum256([]byte(GetConfigurationString(content))))
+ old := conf.Checksum
+ conf.Checksum = checksum
+ switch {
+ case old == "":
+ log.Log(log.Config).Debug("checksum not set in configuration,
calculated and stored",
+ zap.String("checksum", checksum))
+ case old != checksum:
+ log.Log(log.Config).Warn("checksum in configuration incorrect,
overriding with calculated value",
+ zap.String("oldChecksum", old),
zap.String("newChecksum", checksum))
+ }
}
func ParseAndValidateConfig(content []byte) (*SchedulerConfig, error) {
@@ -193,18 +221,41 @@ func ParseAndValidateConfig(content []byte)
(*SchedulerConfig, error) {
return conf, nil
}
+// GetConfigurationString returns the serialised config content without
checksum.
+// The checksum is placed at the start or the end of the config and to avoid
walking a potentially very large
+// config end to end, only the first and the last checksumScanWindow bytes are
scanned for the checksum key.
func GetConfigurationString(requestBytes []byte) string {
- conf := string(requestBytes)
- checksum := "checksum: "
- checksumLength := 64 + len(checksum)
- if strings.Contains(conf, checksum) {
- checksum += strings.Split(conf, checksum)[1]
- checksum = strings.TrimRight(checksum, "\n")
- if len(checksum) > checksumLength {
- checksum = checksum[:checksumLength]
- }
+ length := len(requestBytes)
+ if length == 0 {
+ return ""
+ }
+ key := []byte(checksumKey)
+ // look for the checksum in the tail of the config first, then in the
head, as the Checksum is the last field of the
+ // SchedulerConfig so standard yaml serialisation places it in the tail
of the config
+ tail := max(length-checksumScanWindow, 0)
+ checksumIdx := bytes.Index(requestBytes[tail:], key)
+ if checksumIdx != -1 {
+ checksumIdx += tail
+ } else {
+ head := min(checksumScanWindow, length)
+ checksumIdx = bytes.Index(requestBytes[:head], key)
+ }
+ // no checksum found: the whole content is used to calculate the
checksum
+ if checksumIdx == -1 {
+ return string(requestBytes)
+ }
+ // a checksum is present: use the partitions key to decide whether it
sits before or after the config
+ partitionsIdx := bytes.Index(requestBytes, []byte(partitionsKey))
+ if partitionsIdx == -1 {
+ // no partitions in the config: nothing to calculate a checksum
over
+ return ""
+ }
+ if checksumIdx < partitionsIdx {
+ // checksum stored before the config: the config runs from the
partitions key to the end
+ return string(requestBytes[partitionsIdx:])
}
- return strings.ReplaceAll(conf, checksum, "")
+ // checksum stored after the config: the config runs up to the checksum
line
+ return string(requestBytes[:checksumIdx])
}
// DefaultSchedulerConfig contains the default scheduler configuration; used
if no other is provided
diff --git a/pkg/common/configs/config_test.go
b/pkg/common/configs/config_test.go
index 72eaf31e..0e2b48c3 100644
--- a/pkg/common/configs/config_test.go
+++ b/pkg/common/configs/config_test.go
@@ -1750,18 +1750,21 @@ partitions:
func TestGetConfigurationString(t *testing.T) {
configBytes := []byte(validConf)
checksum := "checksum: " + fmt.Sprintf("%X", sha256.Sum256(configBytes))
+ // the checksum is calculated on the partitions key so the leading
newline should not be part of it
+ confFromPartitions := strings.TrimPrefix(validConf, "\n")
testCases := []struct {
name string
requestBytes []byte
expectedConfig string
}{
{"No checksum", configBytes, validConf},
- {"Checksum at the beginning", []byte(checksum + validConf),
validConf},
+ {"Checksum at the beginning", []byte(checksum + validConf),
confFromPartitions},
+ {"Checksum at the beginning with newline", []byte(checksum +
"\n" + validConf), confFromPartitions},
{"Checksum at the end", []byte(validConf + checksum),
validConf},
- {"Checksum in the middle", []byte(validConf + checksum + "extra
config"), validConf + "extra config"},
+ {"Checksum at the end with newline", []byte(validConf +
checksum + "\n"), validConf},
{"Empty config and checksum", []byte(""), ""},
{"Empty checksum", []byte(validConf + "checksum: "), validConf},
- {"Empty config", []byte("" + checksum), ""},
+ {"Only checksum no partitions", []byte(checksum), ""},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
@@ -1770,6 +1773,47 @@ func TestGetConfigurationString(t *testing.T) {
}
}
+func TestSetChecksum(t *testing.T) {
+ content := []byte(validConf)
+ expected := fmt.Sprintf("%X",
sha256.Sum256([]byte(GetConfigurationString(content))))
+ testCases := []struct {
+ name string
+ old string
+ }{
+ {"missing checksum is set", ""},
+ {"incorrect checksum is overridden", "TEST"},
+ {"correct checksum is kept", expected},
+ }
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ conf := &SchedulerConfig{Checksum: tc.old}
+ SetChecksum(content, conf)
+ assert.Equal(t, expected, conf.Checksum, "checksum not
set to the calculated value")
+ })
+ }
+ // final nil check, must not panic
+ SetChecksum(content, nil)
+}
+
+func TestChecksumSerialisation(t *testing.T) {
+ conf, err := LoadSchedulerConfigFromByteArray([]byte(validConf))
+ assert.NilError(t, err, "unexpected error loading config")
+ // serialise the config the way it is stored in the config map,
partitions first, checksum line last
+ stored, err := yaml.Marshal(conf)
+ assert.NilError(t, err, "unexpected error serialising config")
+ // serialise again with an empty checksum, the checksum line is omitted
+ conf.Checksum = ""
+ noChecksum, err := yaml.Marshal(conf)
+ assert.NilError(t, err, "unexpected error serialising config without
checksum")
+ assert.Equal(t, string(noChecksum), GetConfigurationString(stored),
+ "stripped config does not match the serialisation without a
checksum")
+ // setChecksum check for the generated checksum
+ expected := fmt.Sprintf("%X", sha256.Sum256(noChecksum))
+ SetChecksum(stored, conf)
+ assert.Equal(t, expected, conf.Checksum,
+ "checksum from SetChecksum does not match the manual
recalculation")
+}
+
func prepareUserLimitsConfig(leafQueueMaxApps uint64, leafQueueMaxResource
string) string {
data := `
partitions:
diff --git a/pkg/scheduler/context.go b/pkg/scheduler/context.go
index 402a2c5f..de36655b 100644
--- a/pkg/scheduler/context.go
+++ b/pkg/scheduler/context.go
@@ -229,9 +229,12 @@ func (cc *ClusterContext) processRMConfigUpdateEvent(event
*rmevent.RMConfigUpda
event.Channel <- &rmevent.Result{Succeeded: false, Reason:
err.Error()}
return
}
- // skip update if config has not changed
+ // skip update if the config has not changed: the checksum is
calculated over the config again which lets
+ // us detect a real change even if the config map was updated for an
unrelated reason
oldConf := configs.ConfigContext.Get(cc.policyGroup)
- if conf.Checksum == oldConf.Checksum {
+ if oldConf != nil && conf.Checksum == oldConf.Checksum {
+ log.Log(log.SchedContext).Info("configuration checksum
unchanged, skipping config update",
+ zap.String("rmID", rmID), zap.String("checksum",
conf.Checksum))
event.Channel <- &rmevent.Result{
Succeeded: true,
}
diff --git a/pkg/webservice/dao/config_info.go
b/pkg/webservice/dao/config_info.go
index 39fdcd05..9f92bbe0 100644
--- a/pkg/webservice/dao/config_info.go
+++ b/pkg/webservice/dao/config_info.go
@@ -21,8 +21,10 @@ package dao
import "github.com/apache/yunikorn-core/pkg/common/configs"
type ValidateConfResponse struct {
- Allowed bool `json:"allowed"` // no omitempty, a false value gives a
quick way to understand the result.
- Reason string `json:"reason,omitempty"`
+ Allowed bool `json:"allowed"` // no omitempty, a false value
gives a quick way to understand the result.
+ Reason string `json:"reason,omitempty"`
+ Checksum string `json:"checksum,omitempty"` // checksum calculated
over the submitted configuration, empty when not allowed
+ ChecksumMatch bool `json:"checksumMatch"` // true when the
checksum in the submitted config matched the calculated checksum
}
type ConfigDAOInfo struct {
diff --git a/pkg/webservice/handlers.go b/pkg/webservice/handlers.go
index f07ab098..cb3421b2 100644
--- a/pkg/webservice/handlers.go
+++ b/pkg/webservice/handlers.go
@@ -169,16 +169,24 @@ func validateQueue(queuePath string) error {
func validateConf(w http.ResponseWriter, r *http.Request) {
writeHeaders(w, r.Method)
+ var result dao.ValidateConfResponse
requestBytes, err := io.ReadAll(r.Body)
if err == nil {
- _, err = configs.LoadSchedulerConfigFromByteArray(requestBytes)
+ var conf *configs.SchedulerConfig
+ conf, err = configs.ParseAndValidateConfig(requestBytes)
+ if err == nil {
+ // capture the checksum that was submitted with the
config before it is overridden with the calculated one,
+ // to report whether the submitted checksum was correct
+ submittedChecksum := conf.Checksum
+ configs.SetChecksum(requestBytes, conf)
+ result.Allowed = true
+ result.Checksum = conf.Checksum
+ result.ChecksumMatch = submittedChecksum ==
conf.Checksum
+ }
}
- var result dao.ValidateConfResponse
if err != nil {
result.Allowed = false
result.Reason = err.Error()
- } else {
- result.Allowed = true
}
if err = json.NewEncoder(w).Encode(result); err != nil {
buildJSONErrorResponse(w, err.Error(),
http.StatusInternalServerError)
diff --git a/pkg/webservice/handlers_test.go b/pkg/webservice/handlers_test.go
index c17a39b7..8a7f510f 100644
--- a/pkg/webservice/handlers_test.go
+++ b/pkg/webservice/handlers_test.go
@@ -392,6 +392,39 @@ func TestValidateConf(t *testing.T) {
}
}
+func TestValidateConfChecksum(t *testing.T) {
+ conf, err := configs.LoadSchedulerConfigFromByteArray([]byte(baseConf))
+ assert.NilError(t, err, "unexpected error loading base config")
+ expected := conf.Checksum
+
+ confTests := []struct {
+ name string
+ content string
+ wantAllowed bool
+ wantChecksum string
+ wantChkMatch bool
+ }{
+ {"no checksum in config", baseConf, true, expected, false},
+ {"correct checksum in config", baseConf + "checksum: " +
expected + "\n", true, expected, true},
+ {"incorrect checksum in config", baseConf + "checksum: TEST\n",
true, expected, false},
+ {"invalid config has no checksum", invalidConf, false, "",
false},
+ }
+ for _, test := range confTests {
+ t.Run(test.name, func(t *testing.T) {
+ req, err := http.NewRequest("POST", "",
strings.NewReader(test.content))
+ assert.NilError(t, err, "new http request must not
return an error")
+ resp := &MockResponseWriter{}
+ validateConf(resp, req)
+ var vcr dao.ValidateConfResponse
+ err = json.Unmarshal(resp.outputBytes, &vcr)
+ assert.NilError(t, err, unmarshalError)
+ assert.Equal(t, vcr.Allowed, test.wantAllowed, "allowed
flag incorrect")
+ assert.Equal(t, vcr.Checksum, test.wantChecksum,
"checksum not as expected")
+ assert.Equal(t, vcr.ChecksumMatch, test.wantChkMatch,
"checksum match flag not as expected")
+ })
+ }
+}
+
func TestUserGroupLimits(t *testing.T) {
confTests := []struct {
content string
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
