add pod metrics controller

cmd/controller/main.go

@@ -24,7 +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/metrics"
 	"github.com/awslabs/karpenter/pkg/controllers/node"
 	"github.com/awslabs/karpenter/pkg/controllers/termination"
 	"github.com/awslabs/karpenter/pkg/utils/env"
@@ -93,11 +93,12 @@ func main() {
 		MetricsBindAddress:     fmt.Sprintf(":%d", options.MetricsPort),
 		HealthProbeBindAddress: fmt.Sprintf(":%d", options.HealthProbePort),
 	})
+
 	if err := manager.RegisterControllers(ctx,
 		allocation.NewController(manager.GetClient(), clientSet.CoreV1(), cloudProvider),
 		termination.NewController(ctx, manager.GetClient(), clientSet.CoreV1(), cloudProvider),
 		node.NewController(manager.GetClient()),
-		nodemetrics.NewController(manager.GetClient()),
+		metrics.NewController(manager.GetClient(), cloudProvider),
 	).Start(ctx); err != nil {
 		panic(fmt.Sprintf("Unable to start manager, %s", err.Error()))
 	}

pkg/controllers/allocation/binpacking/packer.go

@@ -40,7 +40,7 @@ var (
 
 	packDuration = prometheus.NewHistogram(
 		prometheus.HistogramOpts{
-			Namespace: metrics.KarpenterNamespace,
+			Namespace: metrics.Namespace,
 			Subsystem: "allocation_controller",
 			Name:      "binpacking_duration_seconds",
 			Help:      "Duration of binpacking process in seconds.",

pkg/controllers/allocation/launcher.go

@@ -105,7 +105,7 @@ func (l *Launcher) bind(ctx context.Context, node *v1.Node, pods []*v1.Pod) (err
 
 var bindTimeHistogram = prometheus.NewHistogram(
 	prometheus.HistogramOpts{
-		Namespace: metrics.KarpenterNamespace,
+		Namespace: metrics.Namespace,
 		Subsystem: "allocation_controller",
 		Name:      "bind_duration_seconds",
 		Help:      "Duration of bind process in seconds. Broken down by result.",

pkg/controllers/allocation/scheduling/scheduler.go

@@ -34,7 +34,7 @@ import (
 
 var schedulingDuration = prometheus.NewHistogramVec(
 	prometheus.HistogramOpts{
-		Namespace: metrics.KarpenterNamespace,
+		Namespace: metrics.Namespace,
 		Subsystem: "allocation_controller",
 		Name:      "scheduling_duration_seconds",
 		Help:      "Duration of scheduling process in seconds. Broken down by provisioner and error.",

pkg/controllers/metrics/common.go

@@ -0,0 +1,54 @@
+/*
+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 metrics
+
+import (
+	"github.com/awslabs/karpenter/pkg/apis/provisioning/v1alpha5"
+	"github.com/awslabs/karpenter/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+	v1 "k8s.io/api/core/v1"
+)
+
+const (
+	controllerName = "Metrics"
+
+	metricSubsystemCapacity = "capacity"
+	metricSubsystemPods     = "pods"
+
+	metricLabelArch         = "arch"
+	metricLabelInstanceType = "instancetype"
+	metricLabelOS           = "os"
+	metricLabelPhase        = "phase"
+	metricLabelProvisioner  = metrics.ProvisionerLabel
+	metricLabelZone         = "zone"
+
+	nodeLabelArch         = v1.LabelArchStable
+	nodeLabelInstanceType = v1.LabelInstanceTypeStable
+	nodeLabelOS           = v1.LabelOSStable
+	nodeLabelZone         = v1.LabelTopologyZone
+
+	nodeConditionTypeReady = v1.NodeReady
+)
+
+var nodeLabelProvisioner = v1alpha5.ProvisionerNameLabelKey
+
+func publishCount(gaugeVec *prometheus.GaugeVec, labels prometheus.Labels, count int) error {
+	gauge, err := gaugeVec.GetMetricWith(labels)
+	if err != nil {
+		return err
+	}
+	gauge.Set(float64(count))
+	return nil
+}

pkg/controllers/metrics/controller.go

@@ -0,0 +1,166 @@
+/*
+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 metrics
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/awslabs/karpenter/pkg/apis/provisioning/v1alpha5"
+	"github.com/awslabs/karpenter/pkg/cloudprovider"
+	"go.uber.org/multierr"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/util/workqueue"
+	"knative.dev/pkg/logging"
+	controllerruntime "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+type Controller struct {
+	CloudProvider cloudprovider.CloudProvider
+	KubeClient    client.Client
+}
+
+func NewController(kubeClient client.Client, cloudProvider cloudprovider.CloudProvider) *Controller {
+	return &Controller{
+		CloudProvider: cloudProvider,
+		KubeClient:    kubeClient,
+	}
+}
+
+func (c *Controller) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
+	loggerName := fmt.Sprintf("%s.provisioner/%s", strings.ToLower(controllerName), req.Name)
+	ctx = logging.WithLogger(ctx, logging.FromContext(ctx).Named(loggerName))
+
+	// Does the provisioner exist?
+	provisioner := &v1alpha5.Provisioner{}
+	if err := c.KubeClient.Get(ctx, req.NamespacedName, provisioner); err != nil {
+		if !errors.IsNotFound(err) {
+			// Unable to determine existence of the provisioner, try again later.
+			return reconcile.Result{}, err
+		}
+
+		// The provisioner has been deleted.
+		return reconcile.Result{}, nil
+	}
+
+	// The provisioner does exist, so update counters.
+	if err := c.updateCounts(ctx, provisioner); err != nil {
+		return reconcile.Result{}, err
+	}
+
+	// Schedule the next run.
+	return reconcile.Result{RequeueAfter: 10 * time.Second}, nil
+}
+
+func (c *Controller) Register(_ context.Context, m manager.Manager) error {
+	return controllerruntime.
+		NewControllerManagedBy(m).
+		Named(controllerName).
+		For(&v1alpha5.Provisioner{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: 10,
+		}).
+		Complete(c)
+}
+
+func (c *Controller) updateCounts(ctx context.Context, provisioner *v1alpha5.Provisioner) error {
+	updateCountFuncs := []func(context.Context, *v1alpha5.Provisioner) error{
+		c.updateNodeCounts,
+		c.updatePodCounts,
+	}
+	updateCountFuncsLen := len(updateCountFuncs)
+	errors := make([]error, updateCountFuncsLen)
+	workqueue.ParallelizeUntil(ctx, updateCountFuncsLen, updateCountFuncsLen, func(index int) {
+		errors[index] = updateCountFuncs[index](ctx, provisioner)
+	})
+
+	return multierr.Combine(errors...)
+}
+
+func (c *Controller) updateNodeCounts(ctx context.Context, provisioner *v1alpha5.Provisioner) error {
+	instanceTypes, err := c.CloudProvider.GetInstanceTypes(ctx, &provisioner.Spec.Constraints)
+	if err != nil {
+		return err
+	}
+
+	archValues := sets.NewString()
+	instanceTypeValues := sets.NewString()
+	osValues := sets.NewString()
+	zoneValues := sets.NewString()
+	for _, instanceType := range instanceTypes {
+		archValues.Insert(instanceType.Architecture())
+		instanceTypeValues.Insert(instanceType.Name())
+		osValues.Insert(instanceType.OperatingSystems().UnsortedList()...)
+		zoneValues.Insert(instanceType.Zones().UnsortedList()...)
+	}
+	knownValuesForNodeLabels := map[string]sets.String{
+		nodeLabelArch:         archValues,
+		nodeLabelInstanceType: instanceTypeValues,
+		nodeLabelOS:           osValues,
+		nodeLabelZone:         zoneValues,
+	}
+
+	return publishNodeCounts(provisioner.Name, knownValuesForNodeLabels, func(matchingLabels client.MatchingLabels, consume nodeListConsumerFunc) error {
+		nodes := v1.NodeList{}
+		if err := c.KubeClient.List(ctx, &nodes, matchingLabels); err != nil {
+			return err
+		}
+		return consume(nodes.Items)
+	})
+}
+
+func (c *Controller) updatePodCounts(ctx context.Context, provisioner *v1alpha5.Provisioner) error {
+	podsForProvisioner, err := c.podsForProvisioner(ctx, provisioner)
+	if err != nil {
+		return err
+	}
+
+	return publishPodCounts(provisioner.Name, podsForProvisioner)
+}
+
+// podsForProvisioner returns a map of slices containing all pods scheduled to nodes in each zone.
+func (c *Controller) podsForProvisioner(ctx context.Context, provisioner *v1alpha5.Provisioner) ([]v1.Pod, error) {
+	// Karpenter does not apply a label, or other marker, to pods.
+
+	results := []v1.Pod{}
+
+	// 1. Fetch all nodes associated with the provisioner.
+	nodeList := v1.NodeList{}
+	withProvisionerName := client.MatchingLabels{nodeLabelProvisioner: provisioner.Name}
+	if err := c.KubeClient.List(ctx, &nodeList, withProvisionerName); err != nil {
+		return nil, err
+	}
+
+	// 2. Get all the pods scheduled to each node.
+	for _, node := range nodeList.Items {
+		podList := v1.PodList{}
+		withNodeName := client.MatchingFields{"spec.nodeName": node.Name}
+		if err := c.KubeClient.List(ctx, &podList, withNodeName); err != nil {
+			return nil, err
+		}
+
+		results = append(results, podList.Items...)
+	}
+
+	return results, nil
+}