Sharding per node

docs/cli-arguments.md

@@ -46,6 +46,7 @@ Usage of ./kube-state-metrics:
       --metric-opt-in-list string                  Comma-separated list of metrics which are opt-in and not enabled by default. This is in addition to the metric allow- and denylists
       --namespaces string                          Comma-separated list of namespaces to be enabled. Defaults to ""
       --namespaces-denylist string                 Comma-separated list of namespaces not to be enabled. If namespaces and namespaces-denylist are both set, only namespaces that are excluded in namespaces-denylist will be used.
+      --nodename string                            Set spec.nodeName=nodeName when watching resources. Only available for resources which support nodeName filter.
       --one_output                                 If true, only write logs to their native severity level (vs also writing to each lower severity level; no effect when -logtostderr=true)
       --pod string                                 Name of the pod that contains the kube-state-metrics container. When set, it is expected that --pod and --pod-namespace are both set. Most likely this should be passed via the downward API. This is used for auto-detecting sharding. If set, this has preference over statically configured sharding. This is experimental, it may be removed without notice.
       --pod-namespace string                       Name of the namespace of the pod specified by --pod. When set, it is expected that --pod and --pod-namespace are both set. Most likely this should be passed via the downward API. This is used for auto-detecting sharding. If set, this has preference over statically configured sharding. This is experimental, it may be removed without notice.

internal/store/builder.go

@@ -63,6 +63,7 @@ type Builder struct {
 	vpaClient                     vpaclientset.Interface
 	namespaces                    options.NamespaceList
 	namespaceFilter               string
+	nodenameFilter                string
 	ctx                           context.Context
 	enabledResources              []string
 	familyGeneratorFilter         generator.FamilyGeneratorFilter
@@ -112,6 +113,16 @@ func (b *Builder) WithNamespaces(n options.NamespaceList, nsFilter string) {
 	b.namespaceFilter = nsFilter
 }
 
+// WithNodename sets the nodename property of a Builder.
+func (b *Builder) WithNodename(nodenameFilter string) {
+	b.nodenameFilter = nodenameFilter
+}
+
+// MergeFieldSelector merges two fieldSelectors using AND operator.
+func (b *Builder) MergeFieldSelector(s1 string, s2 string) (string, error) {
+	return options.MergeFieldSelector(s1, s2)
+}
+
 // WithSharding sets the shard and totalShards property of a Builder.
 func (b *Builder) WithSharding(shard int32, totalShards int) {
 	b.shard = shard
@@ -467,7 +478,13 @@ func (b *Builder) buildStores(
 			familyHeaders,
 			composedMetricGenFuncs,
 		)
-		listWatcher := listWatchFunc(b.kubeClient, v1.NamespaceAll, b.namespaceFilter)
+		merged, err := b.MergeFieldSelector(b.namespaceFilter, b.nodenameFilter)
+		if err != nil {
+			panic(fmt.Sprintf("Failed to merge fieldSelector %s and %s", b.namespaceFilter, b.nodenameFilter))
+		}
+		klog.Infof("FieldSelector is used ", merged)
+		listWatcher := listWatchFunc(b.kubeClient, v1.NamespaceAll, merged)
+		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		return []cache.Store{store}
 	}
@@ -478,7 +495,12 @@ func (b *Builder) buildStores(
 			familyHeaders,
 			composedMetricGenFuncs,
 		)
-		listWatcher := listWatchFunc(b.kubeClient, ns, b.namespaceFilter)
+		merged, err := b.MergeFieldSelector(b.namespaceFilter, b.nodenameFilter)
+		if err != nil {
+			panic(fmt.Sprintf("Failed to merge fieldSelector %s and %s", b.namespaceFilter, b.nodenameFilter))
+		}
+		klog.Infof("FieldSelector is used ", merged)
+		listWatcher := listWatchFunc(b.kubeClient, ns, merged)
 		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		stores = append(stores, store)
 	}
@@ -508,7 +530,12 @@ func (b *Builder) buildCustomResourceStores(resourceName string,
 			familyHeaders,
 			composedMetricGenFuncs,
 		)
-		listWatcher := listWatchFunc(customResourceClient, v1.NamespaceAll, b.namespaceFilter)
+		merged, err := b.MergeFieldSelector(b.namespaceFilter, b.nodenameFilter)
+		if err != nil {
+			panic(fmt.Sprintf("Failed to merge fieldSelector %s and %s", b.namespaceFilter, b.nodenameFilter))
+		}
+		klog.Infof("FieldSelector is used ", merged)
+		listWatcher := listWatchFunc(customResourceClient, v1.NamespaceAll, merged)
 		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		return []cache.Store{store}
 	}
@@ -519,7 +546,12 @@ func (b *Builder) buildCustomResourceStores(resourceName string,
 			familyHeaders,
 			composedMetricGenFuncs,
 		)
-		listWatcher := listWatchFunc(customResourceClient, ns, b.namespaceFilter)
+		merged, err := b.MergeFieldSelector(b.namespaceFilter, b.nodenameFilter)
+		if err != nil {
+			panic(fmt.Sprintf("Failed to merge fieldSelector %s and %s", b.namespaceFilter, b.nodenameFilter))
+		}
+		klog.Infof("FieldSelector is used ", merged)
+		listWatcher := listWatchFunc(customResourceClient, ns, merged)
 		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		stores = append(stores, store)
 	}

pkg/app/server.go

@@ -108,6 +108,8 @@ func RunKubeStateMetrics(ctx context.Context, opts *options.Options, factories .
 	nsFieldSelector := namespaces.GetExcludeNSFieldSelector(opts.NamespacesDenylist)
 	storeBuilder.WithNamespaces(namespaces, nsFieldSelector)
 
+	storeBuilder.WithNodename(opts.Nodename.GetNodenameFieldSelector())
+
 	allowDenyList, err := allowdenylist.New(opts.MetricAllowlist, opts.MetricDenylist)
 	if err != nil {
 		return err

pkg/options/options.go

@@ -39,6 +39,7 @@ type Options struct {
 	Resources            ResourceSet
 	Namespaces           NamespaceList
 	NamespacesDenylist   NamespaceList
+	Nodename             NodenameType
 	Shard                int32
 	TotalShards          int
 	Pod                  string
@@ -103,6 +104,7 @@ func (o *Options) AddFlags() {
 	o.flags.Var(&o.Resources, "resources", fmt.Sprintf("Comma-separated list of Resources to be enabled. Defaults to %q", &DefaultResources))
 	o.flags.Var(&o.Namespaces, "namespaces", fmt.Sprintf("Comma-separated list of namespaces to be enabled. Defaults to %q", &DefaultNamespaces))
 	o.flags.Var(&o.NamespacesDenylist, "namespaces-denylist", "Comma-separated list of namespaces not to be enabled. If namespaces and namespaces-denylist are both set, only namespaces that are excluded in namespaces-denylist will be used.")
+	o.flags.StringVar((*string)(&o.Nodename), "nodename", "", "Set spec.nodeName=nodeName when watching resources. Only available for resources which support nodeName filter.")
 	o.flags.Var(&o.MetricAllowlist, "metric-allowlist", "Comma-separated list of metrics to be exposed. This list comprises of exact metric names and/or regex patterns. The allowlist and denylist are mutually exclusive.")
 	o.flags.Var(&o.MetricDenylist, "metric-denylist", "Comma-separated list of metrics not to be enabled. This list comprises of exact metric names and/or regex patterns. The allowlist and denylist are mutually exclusive.")
 	o.flags.Var(&o.MetricOptInList, "metric-opt-in-list", "Comma-separated list of metrics which are opt-in and not enabled by default. This is in addition to the metric allow- and denylists")

pkg/options/types.go

@@ -104,6 +104,36 @@ func (r *ResourceSet) Type() string {
 	return "string"
 }
 
+// NodenameType represents a nodeName to query from.
+type NodenameType string
+
+// GetNodenameFieldSelector returns a nodename field selector.
+func (n *NodenameType) GetNodenameFieldSelector() string {
+	if string(*n) != "" {
+		return fields.OneTermEqualSelector("spec.nodeName", string(*n)).String()
+	}
+	return fields.Nothing().String()
+}
+
+// MergeFieldSelector returns AND of two field selectors.
+func MergeFieldSelector(s1 string, s2 string) (string, error) {
+	selector1, err := fields.ParseSelector(s1)
+	if err != nil {
+		return fields.Nothing().String(), err
+	}
+	selector2, err := fields.ParseSelector(s2)
+	if err != nil {
+		return fields.Nothing().String(), err
+	}
+	if selector1.Empty() {
+		return selector2.String(), nil
+	}
+	if selector2.Empty() {
+		return selector1.String(), nil
+	}
+	return fields.AndSelectors(selector1, selector2).String(), nil
+}
+
 // NamespaceList represents a list of namespaces to query from.
 type NamespaceList []string
 

pkg/options/types_test.go

@@ -155,6 +155,100 @@ func TestNamespaceList_ExcludeNamespacesFieldSelector(t *testing.T) {
 	}
 }
 
+func TestNodenameFieldSelector(t *testing.T) {
+	tests := []struct {
+		Desc     string
+		Nodename NodenameType
+		Wanted   string
+	}{
+		{
+			Desc:     "empty node name",
+			Nodename: "",
+			Wanted:   "",
+		},
+		{
+			Desc:     "with node name",
+			Nodename: "k8s-node-1",
+			Wanted:   "spec.nodeName=k8s-node-1",
+		},
+	}
+
+	for _, test := range tests {
+		node := test.Nodename
+		actual := node.GetNodenameFieldSelector()
+		if !reflect.DeepEqual(actual, test.Wanted) {
+			t.Errorf("Test error for Desc: %s. Want: %+v. Got: %+v.", test.Desc, test.Wanted, actual)
+		}
+	}
+}
+
+func TestMergeFieldSelector(t *testing.T) {
+	tests := []struct {
+		Desc             string
+		Namespaces       NamespaceList
+		DeniedNamespaces NamespaceList
+		Nodename         NodenameType
+		Wanted           string
+	}{
+		{
+			Desc:             "empty DeniedNamespaces",
+			Namespaces:       NamespaceList{"default", "kube-system"},
+			DeniedNamespaces: NamespaceList{},
+			Nodename:         "",
+			Wanted:           "",
+		},
+		{
+			Desc:             "all DeniedNamespaces",
+			Namespaces:       DefaultNamespaces,
+			DeniedNamespaces: NamespaceList{"some-system"},
+			Nodename:         "",
+			Wanted:           "metadata.namespace!=some-system",
+		},
+		{
+			Desc:             "general case",
+			Namespaces:       DefaultNamespaces,
+			DeniedNamespaces: NamespaceList{"case1-system", "case2-system"},
+			Nodename:         "",
+			Wanted:           "metadata.namespace!=case1-system,metadata.namespace!=case2-system",
+		},
+		{
+			Desc:             "empty DeniedNamespaces",
+			Namespaces:       NamespaceList{"default", "kube-system"},
+			DeniedNamespaces: NamespaceList{},
+			Nodename:         "k8s-node-1",
+			Wanted:           "spec.nodeName=k8s-node-1",
+		},
+		{
+			Desc:             "all DeniedNamespaces",
+			Namespaces:       DefaultNamespaces,
+			DeniedNamespaces: NamespaceList{"some-system"},
+			Nodename:         "k8s-node-1",
+			Wanted:           "metadata.namespace!=some-system,spec.nodeName=k8s-node-1",
+		},
+		{
+			Desc:             "general case",
+			Namespaces:       DefaultNamespaces,
+			DeniedNamespaces: NamespaceList{"case1-system", "case2-system"},
+			Nodename:         "k8s-node-1",
+			Wanted:           "metadata.namespace!=case1-system,metadata.namespace!=case2-system,spec.nodeName=k8s-node-1",
+		},
+	}
+
+	for _, test := range tests {
+		ns := test.Namespaces
+		deniedNS := test.DeniedNamespaces
+		selector1 := ns.GetExcludeNSFieldSelector(deniedNS)
+		selector2 := test.Nodename.GetNodenameFieldSelector()
+		actual, err := MergeFieldSelector(selector1, selector2)
+		if err != nil {
+			t.Errorf("Test error for Desc: %s. Can't merge field selector %v.", test.Desc, err)
+		}
+		if !reflect.DeepEqual(actual, test.Wanted) {
+			t.Errorf("Test error for Desc: %s. Want: %+v. Got: %+v.", test.Desc, test.Wanted, actual)
+		}
+	}
+}
+
 func TestMetricSetSet(t *testing.T) {
 	tests := []struct {
 		Desc   string