add node metrics controller

cmd/controller/main.go

@@ -24,6 +24,7 @@ import (
 	"github.com/awslabs/karpenter/pkg/cloudprovider/registry"
 	"github.com/awslabs/karpenter/pkg/controllers"
 	"github.com/awslabs/karpenter/pkg/controllers/allocation"
+	nodemetrics "github.com/awslabs/karpenter/pkg/controllers/metrics/node"
 	"github.com/awslabs/karpenter/pkg/controllers/node"
 	"github.com/awslabs/karpenter/pkg/controllers/termination"
 	"github.com/awslabs/karpenter/pkg/utils/env"
@@ -96,6 +97,7 @@ func main() {
 		allocation.NewController(manager.GetClient(), clientSet.CoreV1(), cloudProvider),
 		termination.NewController(ctx, manager.GetClient(), clientSet.CoreV1(), cloudProvider),
 		node.NewController(manager.GetClient()),
+		nodemetrics.NewController(manager.GetClient()),
 	).Start(ctx); err != nil {
 		panic(fmt.Sprintf("Unable to start manager, %s", err.Error()))
 	}

pkg/cloudprovider/registry/register.go

@@ -33,7 +33,7 @@ func NewCloudProvider(ctx context.Context, options cloudprovider.Options) cloudp
 // architectures, and validation logic. This operation should only be called
 // once at startup time. Typically, this call is made by NewCloudProvider(), but
 // must be called if the cloud provider is constructed manually (e.g. tests).
-func RegisterOrDie(ctx context.Context,cloudProvider cloudprovider.CloudProvider) {
+func RegisterOrDie(ctx context.Context, cloudProvider cloudprovider.CloudProvider) {
 	zones := map[string]bool{}
 	architectures := map[string]bool{}
 	operatingSystems := map[string]bool{}

pkg/controllers/metrics/node/controller.go

@@ -0,0 +1,121 @@
+/*
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package node
+
+import (
+	"context"
+	"time"
+
+	"github.com/awslabs/karpenter/pkg/apis/provisioning/v1alpha4"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"knative.dev/pkg/logging"
+	controllerruntime "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+const (
+	controllerName  = "NodeMetrics"
+	requeueInterval = 10 * time.Second
+)
+
+type Controller struct {
+	KubeClient client.Client
+}
+
+func NewController(kubeClient client.Client) *Controller {
+	return &Controller{KubeClient: kubeClient}
+}
+
+func (c *Controller) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
+	ctx = logging.WithLogger(ctx, logging.FromContext(ctx).Named(controllerName))
+
+	provisionerName := req.NamespacedName.Name
+
+	// 1. Has the provisioner been deleted?
+	if err := c.provisionerExists(ctx, req); err != nil {
+		if !errors.IsNotFound(err) {
+			// Unable to determine existence of the provisioner, try again later.
+			return reconcile.Result{Requeue: true}, err
+		}
+
+		// The provisioner has been deleted. Reset all the associated counts to zero.
+		if err := publishNodeCountsForProvisioner(provisionerName, consumeZeroNodes); err != nil {
+			// One or more metrics were not zeroed. Try again later.
+			return reconcile.Result{Requeue: true}, err
+		}
+
+		// Since the provisioner is gone, do not requeue.
+		return reconcile.Result{}, nil
+	}
+
+	// 2. Update node counts associated with this provisioner.
+	if err := publishNodeCountsForProvisioner(provisionerName, c.consumeNodesWith(ctx)); err != nil {
+		// An updated value for one or more metrics was not published. Try again later.
+		return reconcile.Result{Requeue: true}, err
+	}
+
+	// 3. Schedule the next run.
+	return reconcile.Result{RequeueAfter: requeueInterval}, nil
+}
+
+func (c *Controller) Register(_ context.Context, m manager.Manager) error {
+	return controllerruntime.
+		NewControllerManagedBy(m).
+		Named(controllerName).
+		For(&v1alpha4.Provisioner{}, builder.WithPredicates(
+			predicate.Funcs{
+				CreateFunc:  func(_ event.CreateEvent) bool { return true },
+				DeleteFunc:  func(_ event.DeleteEvent) bool { return true },
+				UpdateFunc:  func(_ event.UpdateEvent) bool { return false },
+				GenericFunc: func(_ event.GenericEvent) bool { return false },
+			},
+		)).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: 1,
+		}).
+		Complete(c)
+}
+
+// provisionerExists simply attempts to retrieve the provisioner from the Controller's Client
+// and returns any resulting error.
+func (c *Controller) provisionerExists(ctx context.Context, req reconcile.Request) error {
+	provisioner := v1alpha4.Provisioner{}
+	return c.KubeClient.Get(ctx, req.NamespacedName, &provisioner)
+}
+
+// consumeNodesWith will retrieve matching nodes from the Controller's Client then
+// pass the nodes to `consume` and returns any resulting error. If Client returns an error when
+// retrieving nodes then the error is returned without calling `consume`.
+func (c *Controller) consumeNodesWith(ctx context.Context) consumeNodesWithFunc {
+	return func(nodeLabels client.MatchingLabels, consume nodeListConsumerFunc) error {
+		nodes := v1.NodeList{}
+		if err := c.KubeClient.List(ctx, &nodes, nodeLabels); err != nil {
+			return err
+		}
+		return consume(nodes.Items)
+	}
+}
+
+// consumeZeroNodes calls `consume` with an empty slice and returns any resulting error.
+func consumeZeroNodes(_ client.MatchingLabels, consume nodeListConsumerFunc) error {
+	return consume([]v1.Node{})
+}

pkg/controllers/metrics/node/counter.go

@@ -0,0 +1,235 @@
+/*
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package node
+
+import (
+	"strings"
+
+	"github.com/awslabs/karpenter/pkg/apis/provisioning/v1alpha4"
+	"github.com/awslabs/karpenter/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/multierr"
+	v1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
+)
+
+const (
+	metricNamespace = metrics.KarpenterNamespace
+	metricSubsystem = "capacity"
+
+	metricLabelArch         = "arch"
+	metricLabelInstanceType = "instancetype"
+	metricLabelOS           = "os"
+	metricLabelProvisioner  = metrics.ProvisionerLabel
+	metricLabelZone         = "zone"
+
+	nodeLabelArch         = v1.LabelArchStable
+	nodeLabelInstanceType = v1.LabelInstanceTypeStable
+	nodeLabelOS           = v1.LabelOSStable
+	nodeLabelZone         = v1.LabelTopologyZone
+
+	nodeConditionTypeReady = v1.NodeReady
+)
+
+type (
+	nodeListConsumerFunc = func([]v1.Node) error
+	consumeNodesWithFunc = func(client.MatchingLabels, nodeListConsumerFunc) error
+)
+
+var (
+	nodeLabelProvisioner = v1alpha4.ProvisionerNameLabelKey
+
+	knownValuesForNodeLabels = v1alpha4.WellKnownLabels
+
+	nodeCountByProvisioner = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: metricNamespace,
+			Subsystem: metricSubsystem,
+			Name:      "node_count",
+			Help:      "Total node count by provisioner.",
+		},
+		[]string{
+			metricLabelProvisioner,
+		},
+	)
+
+	readyNodeCountByProvisionerZone = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: metricNamespace,
+			Subsystem: metricSubsystem,
+			Name:      "ready_node_count",
+			Help:      "Count of nodes that are ready by provisioner and zone.",
+		},
+		[]string{
+			metricLabelProvisioner,
+			metricLabelZone,
+		},
+	)
+
+	readyNodeCountByArchProvisionerZone = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: metricNamespace,
+			Subsystem: metricSubsystem,
+			Name:      "ready_node_arch_count",
+			Help:      "Count of nodes that are ready by architecture, provisioner, and zone.",
+		},
+		[]string{
+			metricLabelArch,
+			metricLabelProvisioner,
+			metricLabelZone,
+		},
+	)
+
+	readyNodeCountByInstancetypeProvisionerZone = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: metricNamespace,
+			Subsystem: metricSubsystem,
+			Name:      "ready_node_instancetype_count",
+			Help:      "Count of nodes that are ready by instance type, provisioner, and zone.",
+		},
+		[]string{
+			metricLabelInstanceType,
+			metricLabelProvisioner,
+			metricLabelZone,
+		},
+	)
+
+	readyNodeCountByOsProvisionerZone = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: metricNamespace,
+			Subsystem: metricSubsystem,
+			Name:      "ready_node_os_count",
+			Help:      "Count of nodes that are ready by operating system, provisioner, and zone.",
+		},
+		[]string{
+			metricLabelOS,
+			metricLabelProvisioner,
+			metricLabelZone,
+		},
+	)
+)
+
+func init() {
+	crmetrics.Registry.MustRegister(nodeCountByProvisioner)
+	crmetrics.Registry.MustRegister(readyNodeCountByProvisionerZone)
+	crmetrics.Registry.MustRegister(readyNodeCountByArchProvisionerZone)
+	crmetrics.Registry.MustRegister(readyNodeCountByInstancetypeProvisionerZone)
+	crmetrics.Registry.MustRegister(readyNodeCountByOsProvisionerZone)
+}
+
+func publishNodeCountsForProvisioner(provisioner string, consumeNodesWith consumeNodesWithFunc) error {
+	archValues := knownValuesForNodeLabels[nodeLabelArch]
+	instanceTypeValues := knownValuesForNodeLabels[nodeLabelInstanceType]
+	osValues := knownValuesForNodeLabels[nodeLabelOS]
+	zoneValues := knownValuesForNodeLabels[nodeLabelZone]
+
+	errors := make([]error, 0, len(archValues)*len(instanceTypeValues)*len(osValues)*len(zoneValues))
+
+	nodeLabels := client.MatchingLabels{nodeLabelProvisioner: provisioner}
+	errors = append(errors, consumeNodesWith(nodeLabels, func(nodes []v1.Node) error {
+		return publishCount(nodeCountByProvisioner, metricLabelsFrom(nodeLabels), len(nodes))
+	}))
+
+	for _, zone := range zoneValues {
+		nodeLabels = client.MatchingLabels{
+			nodeLabelProvisioner: provisioner,
+			nodeLabelZone:        zone,
+		}
+		errors = append(errors, consumeNodesWith(nodeLabels, filterReadyNodes(func(readyNodes []v1.Node) error {
+			return publishCount(readyNodeCountByProvisionerZone, metricLabelsFrom(nodeLabels), len(readyNodes))
+		})))
+
+		for _, arch := range archValues {
+			nodeLabels := client.MatchingLabels{
+				nodeLabelArch:        arch,
+				nodeLabelProvisioner: provisioner,
+				nodeLabelZone:        zone,
+			}
+			errors = append(errors, consumeNodesWith(nodeLabels, filterReadyNodes(func(readyNodes []v1.Node) error {
+				return publishCount(readyNodeCountByArchProvisionerZone, metricLabelsFrom(nodeLabels), len(readyNodes))
+			})))
+		}
+
+		for _, instanceType := range instanceTypeValues {
+			nodeLabels := client.MatchingLabels{
+				nodeLabelInstanceType: instanceType,
+				nodeLabelProvisioner:  provisioner,
+				nodeLabelZone:         zone,
+			}
+			errors = append(errors, consumeNodesWith(nodeLabels, filterReadyNodes(func(readyNodes []v1.Node) error {
+				return publishCount(readyNodeCountByInstancetypeProvisionerZone, metricLabelsFrom(nodeLabels), len(readyNodes))
+			})))
+		}
+
+		for _, os := range osValues {
+			nodeLabels := client.MatchingLabels{
+				nodeLabelOS:          os,
+				nodeLabelProvisioner: provisioner,
+				nodeLabelZone:        zone,
+			}
+			errors = append(errors, consumeNodesWith(nodeLabels, filterReadyNodes(func(readyNodes []v1.Node) error {
+				return publishCount(readyNodeCountByOsProvisionerZone, metricLabelsFrom(nodeLabels), len(readyNodes))
+			})))
+		}
+	}
+
+	return multierr.Combine(errors...)
+}
+
+// filterReadyNodes returns a new function that will filter "ready" nodes to pass on
+// to `consume`, and returns the result.
+func filterReadyNodes(consume nodeListConsumerFunc) nodeListConsumerFunc {
+	return func(nodes []v1.Node) error {
+		readyNodes := make([]v1.Node, 0, len(nodes))
+		for _, node := range nodes {
+			for _, condition := range node.Status.Conditions {
+				if condition.Type == nodeConditionTypeReady && strings.ToLower(string(condition.Status)) == "true" {
+					readyNodes = append(readyNodes, node)
+				}
+			}
+		}
+		return consume(readyNodes)
+	}
+}
+
+func metricLabelsFrom(nodeLabels client.MatchingLabels) prometheus.Labels {
+	metricLabels := prometheus.Labels{}
+	// Exclude node label values that not present or are empty strings.
+	if arch := nodeLabels[nodeLabelArch]; arch != "" {
+		metricLabels[metricLabelArch] = arch
+	}
+	if instanceType := nodeLabels[nodeLabelInstanceType]; instanceType != "" {
+		metricLabels[metricLabelInstanceType] = instanceType
+	}
+	if os := nodeLabels[nodeLabelOS]; os != "" {
+		metricLabels[metricLabelOS] = os
+	}
+	if provisioner := nodeLabels[nodeLabelProvisioner]; provisioner != "" {
+		metricLabels[metricLabelProvisioner] = provisioner
+	}
+	if zone := nodeLabels[nodeLabelZone]; zone != "" {
+		metricLabels[metricLabelZone] = zone
+	}
+	return metricLabels
+}
+
+func publishCount(gaugeVec *prometheus.GaugeVec, labels prometheus.Labels, count int) error {
+	gauge, err := gaugeVec.GetMetricWith(labels)
+	if err == nil {
+		gauge.Set(float64(count))
+	}
+	return err
+}