Expose an instance count metric (#444)

* Extend the cloud provider interface with GetMetricsForMachines

* Basic plumbing for SetInstanceNumberForMachines

* Implement SetInstanceNumberForMachines for AWS

* Fix linting

* Use prometheus.Default{Registry,Gatherer} to allow providers to register individual metrics

* Change interface

* Properly implement SetMetricsForMachines

* Move polling of SetMetricsForMachines into metrics.go

* Call SetMetricsForMachines for all providers individually

* Make aws metrics for instance be able to cope with multiple credentials

* Rebase fixups

* Do not throw an error when SetMetricsForMachines has no implementation

Author: alvaroaleman

cmd/controller/main.go

@@ -225,7 +225,7 @@ func main() {
 		glog.Fatalf("error building kubernetes clientset for leaderElectionClient: %v", err)
 	}
 
-	prometheusRegistry := prometheus.NewRegistry()
+	prometheusRegistry := prometheus.DefaultRegisterer
 
 	// before we acquire a lock we actually warm up caches mirroring the state of the API server
 	clusterInformerFactory := clusterinformers.NewFilteredSharedInformerFactory(machineClient, time.Minute*15, metav1.NamespaceAll, labelSelector(name))
@@ -281,14 +281,12 @@ func main() {
 	ctx, ctxDone := context.WithCancel(context.Background())
 	var g run.Group
 	{
-		prometheusRegistry.MustRegister(prometheus.NewProcessCollector(os.Getpid(), ""))
-		prometheusRegistry.MustRegister(prometheus.NewGoCollector())
 		prometheusRegistry.MustRegister(machinecontroller.NewMachineCollector(
 			clusterInformerFactory.Cluster().V1alpha1().Machines().Lister(),
 			kubeClient,
 		))
 
-		s := createUtilHTTPServer(kubeClient, kubeconfigProvider, prometheusRegistry)
+		s := createUtilHTTPServer(kubeClient, kubeconfigProvider, prometheus.DefaultGatherer)
 		g.Add(func() error {
 			return s.ListenAndServe()
 		}, func(err error) {

pkg/cloudprovider/cloud/provider.go

@@ -45,11 +45,16 @@ type Provider interface {
 	// about created machines, e.g. instance type, instance size, region
 	// or whatever the provider deems interesting. Should always return
 	// a "size" label.
+	// This should not do any api calls to the cloud provider
 	MachineMetricsLabels(machine *clusterv1alpha1.Machine) (map[string]string, error)
 
 	// MigrateUID is called when the controller migrates types and the UID of the machine object changes
 	// All cloud providers that use Machine.UID to uniquely identify resources must implement this
 	MigrateUID(machine *clusterv1alpha1.Machine, new types.UID) error
+
+	// SetMetricsForMachines allows providers to provide provider-specific metrics. This may be implemented
+	// as no-op
+	SetMetricsForMachines(machines clusterv1alpha1.MachineList) error
 }
 
 // MachineUpdater defines a function to persist an update to a machine

pkg/cloudprovider/provider/aws/provider.go

@@ -8,6 +8,16 @@ import (
 	"sync"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/aws/aws-sdk-go/service/iam"
+	"github.com/golang/glog"
+	gocache "github.com/patrickmn/go-cache"
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/kubermatic/machine-controller/pkg/cloudprovider/cloud"
 	cloudprovidererrors "github.com/kubermatic/machine-controller/pkg/cloudprovider/errors"
 	"github.com/kubermatic/machine-controller/pkg/cloudprovider/instance"
@@ -18,19 +28,23 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/awserr"
-	"github.com/aws/aws-sdk-go/aws/credentials"
-	"github.com/aws/aws-sdk-go/aws/session"
-	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/aws/aws-sdk-go/service/iam"
-	"github.com/golang/glog"
-	gocache "github.com/patrickmn/go-cache"
-
 	"sigs.k8s.io/cluster-api/pkg/apis/cluster/common"
 	"sigs.k8s.io/cluster-api/pkg/apis/cluster/v1alpha1"
 )
 
+var (
+	prometheusRegisterer       = &sync.Once{}
+	metricInstancesForMachines = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "machine_controller_aws_instances_for_machine",
+		Help: "The number of instances at aws for a given machine"}, []string{"machine"})
+)
+
+func init() {
+	prometheusRegisterer.Do(func() {
+		prometheus.MustRegister(metricInstancesForMachines)
+	})
+}
+
 type provider struct {
 	configVarResolver *providerconfig.ConfigVarResolver
 }
@@ -766,3 +780,84 @@ func setProviderSpec(rawConfig RawConfig, s v1alpha1.ProviderSpec) (*runtime.Raw
 
 	return &runtime.RawExtension{Raw: rawPconfig}, nil
 }
+
+func (p *provider) SetMetricsForMachines(machines v1alpha1.MachineList) error {
+	if len(machines.Items) < 1 {
+		return nil
+	}
+
+	type ec2Credentials struct {
+		acccessKeyID    string
+		secretAccessKey string
+		region          string
+	}
+
+	var errors []error
+	credentials := map[string]ec2Credentials{}
+	for _, machine := range machines.Items {
+		config, _, _, err := p.getConfig(machines.Items[0].Spec.ProviderSpec)
+		if err != nil {
+			errors = append(errors, fmt.Errorf("failed to parse MachineSpec of machine %s/%s, due to %v", machine.Namespace, machine.Name, err))
+			continue
+		}
+
+		// Very simple and very stupid
+		credentials[fmt.Sprintf("%s/%s/%s", config.AccessKeyID, config.SecretAccessKey, config.Region)] = ec2Credentials{
+			acccessKeyID:    config.AccessKeyID,
+			secretAccessKey: config.SecretAccessKey,
+			region:          config.Region,
+		}
+
+	}
+
+	allReservations := []*ec2.Reservation{}
+	for _, cred := range credentials {
+		ec2Client, err := getEC2client(cred.acccessKeyID, cred.secretAccessKey, cred.region)
+		if err != nil {
+			errors = append(errors, fmt.Errorf("failed to get EC2 client: %v", err))
+			continue
+		}
+		inOut, err := ec2Client.DescribeInstances(&ec2.DescribeInstancesInput{})
+		if err != nil {
+			errors = append(errors, fmt.Errorf("failed to get EC2 instances: %v", err))
+			continue
+		}
+		allReservations = append(allReservations, inOut.Reservations...)
+	}
+
+	for _, machine := range machines.Items {
+		metricInstancesForMachines.WithLabelValues(fmt.Sprintf("%s/%s", machine.Namespace, machine.Name)).Set(
+			getIntanceCountForMachine(machine, allReservations))
+	}
+
+	if len(errors) > 0 {
+		return fmt.Errorf("errors: %v", errors)
+	}
+
+	return nil
+}
+
+func getIntanceCountForMachine(machine v1alpha1.Machine, reservations []*ec2.Reservation) float64 {
+	var count float64
+	for _, reservation := range reservations {
+		for _, i := range reservation.Instances {
+			if i.State == nil ||
+				i.State.Name == nil ||
+				*i.State.Name == ec2.InstanceStateNameTerminated {
+				continue
+			}
+
+			for _, tag := range i.Tags {
+				if *tag.Key != machineUIDTag {
+					continue
+				}
+
+				if *tag.Value == string(machine.UID) {
+					count = count + 1
+				}
+				break
+			}
+		}
+	}
+	return count
+}

pkg/cloudprovider/provider/azure/provider.go

@@ -819,3 +819,7 @@ func (p *provider) MachineMetricsLabels(machine *v1alpha1.Machine) (map[string]s
 
 	return labels, err
 }
+
+func (p *provider) SetMetricsForMachines(machines v1alpha1.MachineList) error {
+	return nil
+}

pkg/cloudprovider/provider/digitalocean/provider.go

@@ -495,3 +495,7 @@ func doStatusAndErrToTerminalError(status int, err error) error {
 		return err
 	}
 }
+
+func (p *provider) SetMetricsForMachines(machines v1alpha1.MachineList) error {
+	return nil
+}

pkg/cloudprovider/provider/fake/provider.go

@@ -91,3 +91,7 @@ func (p *provider) MigrateUID(machine *v1alpha1.Machine, new types.UID) error {
 func (p *provider) MachineMetricsLabels(machine *v1alpha1.Machine) (map[string]string, error) {
 	return map[string]string{}, nil
 }
+
+func (p *provider) SetMetricsForMachines(_ v1alpha1.MachineList) error {
+	return nil
+}

pkg/cloudprovider/provider/hetzner/provider.go

@@ -401,3 +401,7 @@ func hzErrorToTerminalError(err error, msg string) error {
 
 	return err
 }
+
+func (p *provider) SetMetricsForMachines(machines v1alpha1.MachineList) error {
+	return nil
+}

pkg/cloudprovider/provider/kubevirt/provider.go

@@ -15,10 +15,10 @@ import (
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 
-	"k8s.io/apimachinery/pkg/types"
 	kubevirtv1 "kubevirt.io/kubevirt/pkg/api/v1"
 	"kubevirt.io/kubevirt/pkg/kubecli"
 
@@ -353,3 +353,7 @@ func parseResources(cpus, memory string) (*corev1.ResourceList, error) {
 		corev1.ResourceCPU:    cpuResource,
 	}, nil
 }
+
+func (p *provider) SetMetricsForMachines(machines v1alpha1.MachineList) error {
+	return nil
+}

pkg/cloudprovider/provider/openstack/provider.go

@@ -861,3 +861,7 @@ func assignFloatingIPToInstance(machineUpdater cloud.MachineUpdater, machine *v1
 	glog.V(2).Infof("Successfully assigned the FloatingIP %s to instance %s. Took %f seconds(without the recheck wait period %f seconds). ", ip.FloatingIP, instanceID, secondsTook, floatingReassignIPCheckPeriod.Seconds())
 	return nil
 }
+
+func (p *provider) SetMetricsForMachines(machines v1alpha1.MachineList) error {
+	return nil
+}

pkg/cloudprovider/provider/vsphere/provider.go

@@ -636,3 +636,7 @@ func (p *provider) MachineMetricsLabels(machine *v1alpha1.Machine) (map[string]s
 
 	return labels, err
 }
+
+func (p *provider) SetMetricsForMachines(machines v1alpha1.MachineList) error {
+	return nil
+}