Merge pull request #1864 from CatherineF-dev/sharding-per-node

Sharding metrics per node via fieldSelector

Author: k8s-ci-robot

README.md

@@ -45,6 +45,7 @@ are deleted they are no longer visible on the `/metrics` endpoint.
   - [Resource recommendation](#resource-recommendation)
   - [Horizontal sharding](#horizontal-sharding)
     - [Automated sharding](#automated-sharding)
+  - [Daemonset sharding for pod metrics](#daemonset-sharding-pod-metrics)
 - [Setup](#setup)
   - [Building the Docker container](#building-the-docker-container)
 - [Usage](#usage)
@@ -235,6 +236,36 @@ This way of deploying shards is useful when you want to manage KSM shards throug
 
 The downside of using an auto-sharded setup comes from the rollout strategy supported by `StatefulSet`s. When managed by a `StatefulSet`, pods are replaced one at a time with each pod first getting terminated and then recreated. Besides such rollouts being slower, they will also lead to short downtime for each shard. If a Prometheus scrape happens during a rollout, it can miss some of the metrics exported by kube-state-metrics.
 
+### Daemonset sharding for pod metrics
+
+For pod metrics, they can be sharded per node with the following flag:
+* `--node`
+
+Each kube-state-metrics pod uses FieldSelector (spec.nodeName) to watch/list pod metrics only on the same node.
+
+A daemonset kube-state-metrics example:
+```
+apiVersion: apps/v1
+kind: DaemonSet
+spec:
+  template:
+    spec:
+      containers:
+      - image: registry.k8s.io/kube-state-metrics/kube-state-metrics:IMAGE_TAG
+        name: kube-state-metrics
+        args:
+        - --resource=pods
+        - --node=$(NODE_NAME)
+        env:
+        - name: NODE_NAME
+          valueFrom:
+            fieldRef:
+              apiVersion: v1
+              fieldPath: spec.nodeName
+```
+
+Other metrics can be sharded via [Horizontal sharding](#horizontal-sharding).
+
 ### Setup
 
 Install this project to your `$GOPATH` using `go get`:

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.
+      --node string                                Name of the node that contains the kube-state-metrics pod. Most likely it should be passed via the downward API. This is used for daemonset sharding. Only available for resources (pod metrics) that support spec.nodeName fieldSelector. This is experimental.
       --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

@@ -58,11 +58,12 @@ var _ ksmtypes.BuilderInterface = &Builder{}
 // Builder helps to build store. It follows the builder pattern
 // (https://en.wikipedia.org/wiki/Builder_pattern).
 type Builder struct {
-	kubeClient                    clientset.Interface
-	customResourceClients         map[string]interface{}
-	vpaClient                     vpaclientset.Interface
-	namespaces                    options.NamespaceList
-	namespaceFilter               string
+	kubeClient            clientset.Interface
+	customResourceClients map[string]interface{}
+	vpaClient             vpaclientset.Interface
+	namespaces            options.NamespaceList
+	// namespaceFilter is inside fieldSelectorFilter
+	fieldSelectorFilter           string
 	ctx                           context.Context
 	enabledResources              []string
 	familyGeneratorFilter         generator.FamilyGeneratorFilter
@@ -106,10 +107,19 @@ func (b *Builder) WithEnabledResources(r []string) error {
 	return nil
 }
 
+// WithFieldSelectorFilter sets the fieldSelector property of a Builder.
+func (b *Builder) WithFieldSelectorFilter(fieldSelectorFilter string) {
+	b.fieldSelectorFilter = fieldSelectorFilter
+}
+
 // WithNamespaces sets the namespaces property of a Builder.
-func (b *Builder) WithNamespaces(n options.NamespaceList, nsFilter string) {
+func (b *Builder) WithNamespaces(n options.NamespaceList) {
 	b.namespaces = n
-	b.namespaceFilter = nsFilter
+}
+
+// MergeFieldSelectors merges multiple fieldSelectors using AND operator.
+func (b *Builder) MergeFieldSelectors(selectors []string) (string, error) {
+	return options.MergeFieldSelectors(selectors)
 }
 
 // WithSharding sets the shard and totalShards property of a Builder.
@@ -467,7 +477,10 @@ func (b *Builder) buildStores(
 			familyHeaders,
 			composedMetricGenFuncs,
 		)
-		listWatcher := listWatchFunc(b.kubeClient, v1.NamespaceAll, b.namespaceFilter)
+		if b.fieldSelectorFilter != "" {
+			klog.Infof("FieldSelector is used %s", b.fieldSelectorFilter)
+		}
+		listWatcher := listWatchFunc(b.kubeClient, v1.NamespaceAll, b.fieldSelectorFilter)
 		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		return []cache.Store{store}
 	}
@@ -478,7 +491,10 @@ func (b *Builder) buildStores(
 			familyHeaders,
 			composedMetricGenFuncs,
 		)
-		listWatcher := listWatchFunc(b.kubeClient, ns, b.namespaceFilter)
+		if b.fieldSelectorFilter != "" {
+			klog.Infof("FieldSelector is used %s", b.fieldSelectorFilter)
+		}
+		listWatcher := listWatchFunc(b.kubeClient, ns, b.fieldSelectorFilter)
 		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		stores = append(stores, store)
 	}
@@ -508,7 +524,10 @@ func (b *Builder) buildCustomResourceStores(resourceName string,
 			familyHeaders,
 			composedMetricGenFuncs,
 		)
-		listWatcher := listWatchFunc(customResourceClient, v1.NamespaceAll, b.namespaceFilter)
+		if b.fieldSelectorFilter != "" {
+			klog.Infof("FieldSelector is used %s", b.fieldSelectorFilter)
+		}
+		listWatcher := listWatchFunc(customResourceClient, v1.NamespaceAll, b.fieldSelectorFilter)
 		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		return []cache.Store{store}
 	}
@@ -519,7 +538,8 @@ func (b *Builder) buildCustomResourceStores(resourceName string,
 			familyHeaders,
 			composedMetricGenFuncs,
 		)
-		listWatcher := listWatchFunc(customResourceClient, ns, b.namespaceFilter)
+		klog.Infof("FieldSelector is used %s", b.fieldSelectorFilter)
+		listWatcher := listWatchFunc(customResourceClient, ns, b.fieldSelectorFilter)
 		b.startReflector(expectedType, store, listWatcher, useAPIServerCache)
 		stores = append(stores, store)
 	}

main.go

@@ -53,6 +53,11 @@ func main() {
 		os.Exit(0)
 	}
 
+	if err := opts.Validate(); err != nil {
+		klog.ErrorS(err, "Validating options error")
+		klog.FlushAndExit(klog.ExitFlushTimeout, 1)
+	}
+
 	var factories []customresource.RegistryFactory
 	if config, set := resolveCustomResourceConfig(opts); set {
 		crf, err := customresourcestate.FromConfig(config)

pkg/app/server.go

@@ -106,7 +106,13 @@ func RunKubeStateMetrics(ctx context.Context, opts *options.Options, factories .
 
 	namespaces := opts.Namespaces.GetNamespaces()
 	nsFieldSelector := namespaces.GetExcludeNSFieldSelector(opts.NamespacesDenylist)
-	storeBuilder.WithNamespaces(namespaces, nsFieldSelector)
+	nodeFieldSelector := opts.Node.GetNodeFieldSelector()
+	merged, err := storeBuilder.MergeFieldSelectors([]string{nsFieldSelector, nodeFieldSelector})
+	if err != nil {
+		return err
+	}
+	storeBuilder.WithNamespaces(namespaces)
+	storeBuilder.WithFieldSelectorFilter(merged)
 
 	allowDenyList, err := allowdenylist.New(opts.MetricAllowlist, opts.MetricDenylist)
 	if err != nil {

pkg/app/server_test.go

@@ -80,7 +80,7 @@ func BenchmarkKubeStateMetrics(b *testing.B) {
 	builder.WithKubeClient(kubeClient)
 	builder.WithSharding(0, 1)
 	builder.WithContext(ctx)
-	builder.WithNamespaces(options.DefaultNamespaces, "")
+	builder.WithNamespaces(options.DefaultNamespaces)
 	builder.WithGenerateStoresFunc(builder.DefaultGenerateStoresFunc())
 
 	allowDenyListFilter, err := allowdenylist.New(map[string]struct{}{}, map[string]struct{}{})
@@ -155,7 +155,7 @@ func TestFullScrapeCycle(t *testing.T) {
 		t.Fatal(err)
 	}
 	builder.WithKubeClient(kubeClient)
-	builder.WithNamespaces(options.DefaultNamespaces, "")
+	builder.WithNamespaces(options.DefaultNamespaces)
 	builder.WithGenerateStoresFunc(builder.DefaultGenerateStoresFunc())
 
 	l, err := allowdenylist.New(map[string]struct{}{}, map[string]struct{}{})
@@ -442,7 +442,7 @@ func TestShardingEquivalenceScrapeCycle(t *testing.T) {
 		t.Fatal(err)
 	}
 	unshardedBuilder.WithKubeClient(kubeClient)
-	unshardedBuilder.WithNamespaces(options.DefaultNamespaces, "")
+	unshardedBuilder.WithNamespaces(options.DefaultNamespaces)
 	unshardedBuilder.WithFamilyGeneratorFilter(l)
 	unshardedBuilder.WithAllowLabels(map[string][]string{})
 	unshardedBuilder.WithGenerateStoresFunc(unshardedBuilder.DefaultGenerateStoresFunc())
@@ -458,7 +458,7 @@ func TestShardingEquivalenceScrapeCycle(t *testing.T) {
 		t.Fatal(err)
 	}
 	shardedBuilder1.WithKubeClient(kubeClient)
-	shardedBuilder1.WithNamespaces(options.DefaultNamespaces, "")
+	shardedBuilder1.WithNamespaces(options.DefaultNamespaces)
 	shardedBuilder1.WithFamilyGeneratorFilter(l)
 	shardedBuilder1.WithAllowLabels(map[string][]string{})
 	shardedBuilder1.WithGenerateStoresFunc(shardedBuilder1.DefaultGenerateStoresFunc())
@@ -474,7 +474,7 @@ func TestShardingEquivalenceScrapeCycle(t *testing.T) {
 		t.Fatal(err)
 	}
 	shardedBuilder2.WithKubeClient(kubeClient)
-	shardedBuilder2.WithNamespaces(options.DefaultNamespaces, "")
+	shardedBuilder2.WithNamespaces(options.DefaultNamespaces)
 	shardedBuilder2.WithFamilyGeneratorFilter(l)
 	shardedBuilder2.WithAllowLabels(map[string][]string{})
 	shardedBuilder2.WithGenerateStoresFunc(shardedBuilder2.DefaultGenerateStoresFunc())
@@ -616,7 +616,7 @@ func TestCustomResourceExtension(t *testing.T) {
 
 	builder.WithKubeClient(kubeClient)
 	builder.WithCustomResourceClients(customResourceClients)
-	builder.WithNamespaces(options.DefaultNamespaces, "")
+	builder.WithNamespaces(options.DefaultNamespaces)
 	builder.WithGenerateStoresFunc(builder.DefaultGenerateStoresFunc())
 	builder.WithGenerateCustomResourceStoresFunc(builder.DefaultGenerateCustomResourceStoresFunc())
 

pkg/builder/builder.go

@@ -58,8 +58,8 @@ func (b *Builder) WithEnabledResources(c []string) error {
 }
 
 // WithNamespaces sets the namespaces property of a Builder.
-func (b *Builder) WithNamespaces(n options.NamespaceList, nsFilter string) {
-	b.internal.WithNamespaces(n, nsFilter)
+func (b *Builder) WithNamespaces(n options.NamespaceList) {
+	b.internal.WithNamespaces(n)
 }
 
 // WithSharding sets the shard and totalShards property of a Builder.

pkg/builder/types/interfaces.go

@@ -35,7 +35,8 @@ import (
 type BuilderInterface interface {
 	WithMetrics(r prometheus.Registerer)
 	WithEnabledResources(c []string) error
-	WithNamespaces(n options.NamespaceList, nsFilter string)
+	WithNamespaces(n options.NamespaceList)
+	WithFieldSelectorFilter(fieldSelectors string)
 	WithSharding(shard int32, totalShards int)
 	WithContext(ctx context.Context)
 	WithKubeClient(c clientset.Interface)

pkg/options/options.go

@@ -39,6 +39,7 @@ type Options struct {
 	Resources            ResourceSet
 	Namespaces           NamespaceList
 	NamespacesDenylist   NamespaceList
+	Node                 NodeType
 	Shard                int32
 	TotalShards          int
 	Pod                  string
@@ -111,6 +112,8 @@ func (o *Options) AddFlags() {
 	o.flags.Int32Var(&o.Shard, "shard", int32(0), "The instances shard nominal (zero indexed) within the total number of shards. (default 0)")
 	o.flags.IntVar(&o.TotalShards, "total-shards", 1, "The total number of shards. Sharding is disabled when total shards is set to 1.")
 
+	o.flags.StringVar((*string)(&o.Node), "node", "", "Name of the node that contains the kube-state-metrics pod. Most likely it should be passed via the downward API. This is used for daemonset sharding. Only available for resources (pod metrics) that support spec.nodeName fieldSelector. This is experimental.")
+
 	autoshardingNotice := "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."
 
 	o.flags.StringVar(&o.Pod, "pod", "", "Name of the pod that contains the kube-state-metrics container. "+autoshardingNotice)
@@ -132,3 +135,17 @@ func (o *Options) Parse() error {
 func (o *Options) Usage() {
 	o.flags.Usage()
 }
+
+// Validate validates arguments
+func (o *Options) Validate() error {
+	shardableResource := "pods"
+	if o.Node == "" {
+		return nil
+	}
+	for _, x := range o.Resources.AsSlice() {
+		if x != shardableResource {
+			return fmt.Errorf("Resource %s can't be sharded by field selector spec.nodeName", x)
+		}
+	}
+	return nil
+}

pkg/options/types.go

@@ -104,6 +104,54 @@ func (r *ResourceSet) Type() string {
 	return "string"
 }
 
+// NodeType represents a nodeName to query from.
+type NodeType string
+
+// GetNodeFieldSelector returns a nodename field selector.
+func (n *NodeType) GetNodeFieldSelector() string {
+	if string(*n) != "" {
+		return fields.OneTermEqualSelector("spec.nodeName", string(*n)).String()
+	}
+	return EmptyFieldSelector()
+}
+
+// EmptyFieldSelector returns an empty field selector.
+func EmptyFieldSelector() string {
+	return fields.Nothing().String()
+}
+
+// MergeFieldSelectors returns AND of a list of field selectors.
+func MergeFieldSelectors(selectors []string) (string, error) {
+	var err error
+	merged := EmptyFieldSelector()
+	for _, s := range selectors {
+		merged, err = MergeTwoFieldSelectors(merged, s)
+		if err != nil {
+			return "", err
+		}
+	}
+	return merged, nil
+}
+
+// MergeTwoFieldSelectors returns AND of two field selectors.
+func MergeTwoFieldSelectors(s1 string, s2 string) (string, error) {
+	selector1, err := fields.ParseSelector(s1)
+	if err != nil {
+		return EmptyFieldSelector(), err
+	}
+	selector2, err := fields.ParseSelector(s2)
+	if err != nil {
+		return EmptyFieldSelector(), 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