support select container by label

meta.go

@@ -71,14 +71,16 @@ type Resource interface {
 	Stale(other Resource) bool
 	Changed(other Resource) bool
 	Clone() Resource
+	Labels() []Pair
 
 	Marshal() ([]byte, error)
 	Unmarshal(data []byte) error
 }
 
 // Pair key value pair
 type Pair struct {
-	Key, Value string
+	Key   string `json:"key"`
+	Value string `json:"value"`
 }
 
 // Container is the resource container, the resource is running on the container
@@ -90,7 +92,6 @@ type Container interface {
 	ID() uint64
 	Labels() []Pair
 	State() State
-
 	Clone() Container
 
 	Marshal() ([]byte, error)

scheduler_filter.go

@@ -26,6 +26,10 @@ func filterTarget(container *ContainerRuntime, filters []Filter) bool {
 	return false
 }
 
+type labelFilter struct {
+	labels []Pair
+}
+
 type stateFilter struct {
 	cfg *Cfg
 }
@@ -112,6 +116,11 @@ func NewSnapshotCountFilter(cfg *Cfg) Filter {
 	return &snapshotCountFilter{cfg: cfg}
 }
 
+// NewLabelFilter returns label filter, if the container missing the labels, skip it.
+func NewLabelFilter(labels []Pair) Filter {
+	return &labelFilter{labels: labels}
+}
+
 func (f *stateFilter) filter(container *ContainerRuntime) bool {
 	return !(container.IsUp() && container.Downtime() < f.cfg.MaxAllowContainerDownDuration)
 }
@@ -198,3 +207,21 @@ func (f *snapshotCountFilter) FilterSource(container *ContainerRuntime) bool {
 func (f *snapshotCountFilter) FilterTarget(container *ContainerRuntime) bool {
 	return f.filter(container)
 }
+
+func (f *labelFilter) filter(container *ContainerRuntime) bool {
+	for _, l := range f.labels {
+		if l.Value != container.GetLabelValue(l.Key) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (f *labelFilter) FilterSource(container *ContainerRuntime) bool {
+	return f.filter(container)
+}
+
+func (f *labelFilter) FilterTarget(container *ContainerRuntime) bool {
+	return f.filter(container)
+}

scheduler_replica.go

@@ -121,6 +121,7 @@ func (r *replicaChecker) selectBestPeer(target *ResourceRuntime, allocPeerID boo
 	filters = append(filters, NewStateFilter(r.cfg))
 	filters = append(filters, NewStorageThresholdFilter(r.cfg))
 	filters = append(filters, NewExcludedFilter(nil, target.GetContainerIDs()))
+	filters = append(filters, NewLabelFilter(target.meta.Labels())) // make sure the container has the tags needed for the resource
 
 	var (
 		bestContainer *ContainerRuntime