FIX: unique placement.

Author: andreaspeters

.version.json

@@ -1 +1 @@
-[{ "version":"v1.0.3", "builddate":"20240718" }]
+[{ "version":"v1.0.3", "builddate":"20240802" }]

changelog.md

@@ -11,6 +11,7 @@
 - ADD: Kafka Plugin to forware mesos event messages to kafka
 - CHANGE: Migrate to google protobuf
 - FIX: Scale up/down of a mesos task.
+- FIX: unique placement
 
 ## 1.0.3
 

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/Showmax/go-fqdn v1.0.0
 	github.com/confluentinc/confluent-kafka-go v1.9.2
 	github.com/gorilla/mux v1.8.1
-	github.com/redis/go-redis/v9 v9.5.4
+	github.com/redis/go-redis/v9 v9.6.1
 	github.com/sirupsen/logrus v1.9.3
 	google.golang.org/protobuf v1.34.2
 	gopkg.in/yaml.v3 v3.0.1
@@ -18,7 +18,7 @@ require (
 	github.com/cenkalti/backoff/v3 v3.2.2 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
-	github.com/go-jose/go-jose/v4 v4.0.3 // indirect
+	github.com/go-jose/go-jose/v4 v4.0.4 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect

go.sum

@@ -53,8 +53,8 @@ github.com/frankban/quicktest v1.7.2/go.mod h1:jaStnuzAqU1AJdCO0l53JDCJrVDKcS03D
 github.com/frankban/quicktest v1.10.0/go.mod h1:ui7WezCLWMWxVWr1GETZY3smRy0G4KWq9vcPtJmFl7Y=
 github.com/frankban/quicktest v1.14.0/go.mod h1:NeW+ay9A/U67EYXNFA1nPE8e/tnQv/09mUdL/ijj8og=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
-github.com/go-jose/go-jose/v4 v4.0.3 h1:o8aphO8Hv6RPmH+GfzVuyf7YXSBibp+8YyHdOoDESGo=
-github.com/go-jose/go-jose/v4 v4.0.3/go.mod h1:NKb5HO1EZccyMpiZNbdUw/14tiXNyUJh188dfnMCAfc=
+github.com/go-jose/go-jose/v4 v4.0.4 h1:VsjPI33J0SB9vQM6PLmNjoHqMQNGPiZ0rHL7Ni7Q6/E=
+github.com/go-jose/go-jose/v4 v4.0.4/go.mod h1:NKb5HO1EZccyMpiZNbdUw/14tiXNyUJh188dfnMCAfc=
 github.com/go-test/deep v1.0.2 h1:onZX1rnHT3Wv6cqNgYyFOOlgVKJrksuCMCRvJStbMYw=
 github.com/go-test/deep v1.0.2/go.mod h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
@@ -160,8 +160,8 @@ github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsK
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/redis/go-redis/v9 v9.5.4 h1:vOFYDKKVgrI5u++QvnMT7DksSMYg7Aw/Np4vLJLKLwY=
-github.com/redis/go-redis/v9 v9.5.4/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
+github.com/redis/go-redis/v9 v9.6.1 h1:HHDteefn6ZkTtY5fGUE8tj8uy85AHk6zP7CpzIAM0y4=
+github.com/redis/go-redis/v9 v9.6.1/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
 github.com/rogpeppe/clock v0.0.0-20190514195947-2896927a307a/go.mod h1:4r5QyqhjIWCcK8DO4KMclc5Iknq5qVBAlbYYzAbUScQ=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=

redis/redis.go

@@ -125,6 +125,31 @@ func (e *Redis) CountRedisKey(pattern string, ignoreState string) int {
 	return count
 }
 
+// CountRedisKeyState will get back the amount of redis keys with the spesific state.
+func (e *Redis) CountRedisKeyState(pattern string, state string) int {
+	keys := e.GetAllRedisKeys(pattern)
+	count := 0
+	for keys.Next(e.CTX) {
+		// ignore redis keys if they are not mesos tasks
+		if e.CheckIfNotTask(keys) {
+			continue
+		}
+
+		// do not count these key
+		if state != "" {
+			// get the values of the current key
+			key := e.GetRedisKey(keys.Val())
+			task := e.Mesos.DecodeTask(key)
+
+			if task.State != state {
+				continue
+			}
+		}
+		count++
+	}
+	return count
+}
+
 // SaveConfig store the current framework config
 func (e *Redis) SaveConfig(config cfg.Config) {
 	data, _ := json.Marshal(config)

sbom.json

@@ -40,6 +40,12 @@ func (e *Scheduler) Heartbeat() {
 
 		// there are lesser instances are running as it should be
 		if e.Redis.CountRedisKey(task.TaskName+":*", "__KILL") < task.Instances {
+			// do not schedule task if the placement is unique and the amount of running tasks
+			// are same like the amount of mesos agents.
+			if e.getLabelValue("__mc_placement", task) == "unique" && e.Redis.CountRedisKeyState(task.TaskName+":*", "TASK_RUNNING") >= e.Mesos.CountAgent {
+				continue
+			}
+
 			logrus.WithField("func", "scheduler.CheckState").Info("Scale up Mesos Task: ", task.TaskName)
 			e.Mesos.Revive()
 			task.State = ""