Merge branch 'master' into refac

integrations/apis/flux.weave.works/v1beta1/types.go

@@ -1,6 +1,7 @@
 package v1beta1
 
 import (
+	"fmt"
 	"strings"
 
 	"github.com/ghodss/yaml"
@@ -29,6 +30,21 @@ func (fhr HelmRelease) ResourceID() resource.ID {
 	return resource.MakeID(fhr.Namespace, "HelmRelease", fhr.Name)
 }
 
+// ReleaseName returns the configured release name, or constructs and
+// returns one based on the namespace and name of the HelmRelease.
+func (fhr HelmRelease) ReleaseName() string {
+	namespace := fhr.Namespace
+	if namespace == "" {
+		namespace = "default"
+	}
+	releaseName := fhr.Spec.ReleaseName
+	if releaseName == "" {
+		releaseName = fmt.Sprintf("%s-%s", namespace, fhr.Name)
+	}
+
+	return releaseName
+}
+
 // ValuesFromSource represents a source of values.
 // Only one of its fields may be set.
 type ValuesFromSource struct {

integrations/helm/chartsync/chartsync.go

@@ -198,7 +198,7 @@ func (chs *ChartChangeSync) Run(stopCh <-chan struct{}, errc chan error, wg *syn
 					for _, fhr := range resources {
 						ref := fhr.Spec.ChartSource.GitChartSource.RefOrDefault()
 						path := fhr.Spec.ChartSource.GitChartSource.Path
-						releaseName := release.GetReleaseName(fhr)
+						releaseName := fhr.ReleaseName()
 
 						ctx, cancel := context.WithTimeout(context.Background(), helmop.GitOperationTimeout)
 						refHead, err := repo.Revision(ctx, ref)
@@ -291,7 +291,7 @@ func (chs *ChartChangeSync) ReconcileReleaseDef(fhr fluxv1beta1.HelmRelease) {
 // HelmRelease resource, and either installs, upgrades, or does
 // nothing, depending on the state (or absence) of the release.
 func (chs *ChartChangeSync) reconcileReleaseDef(fhr fluxv1beta1.HelmRelease) {
-	releaseName := release.GetReleaseName(fhr)
+	releaseName := fhr.ReleaseName()
 
 	// Attempt to retrieve an upgradable release, in case no release
 	// or error is returned, install it.
@@ -423,7 +423,7 @@ func (chs *ChartChangeSync) reconcileReleaseDef(fhr fluxv1beta1.HelmRelease) {
 // call it when it is handling a resource deletion.
 func (chs *ChartChangeSync) DeleteRelease(fhr fluxv1beta1.HelmRelease) {
 	// FIXME(michael): these may need to stop mirroring a repo.
-	name := release.GetReleaseName(fhr)
+	name := fhr.ReleaseName()
 	err := chs.release.Delete(name)
 	if err != nil {
 		chs.logger.Log("warning", "chart release not deleted", "resource", fhr.ResourceID().String(), "release", name, "err", err)

integrations/helm/operator/metrics.go

@@ -0,0 +1,15 @@
+package operator
+
+import (
+	"github.com/go-kit/kit/metrics/prometheus"
+	stdprometheus "github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	releaseQueueLength = prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+		Namespace: "flux",
+		Subsystem: "helm_operator",
+		Name:      "release_queue_length_count",
+		Help:      "Count of releases waiting in the queue to be processed.",
+	}, []string{})
+)

integrations/helm/operator/operator.go

@@ -149,6 +149,8 @@ func (c *Controller) runWorker() {
 // processNextWorkItem will read a single work item off the workqueue and
 // attempt to process it, by calling the syncHandler.
 func (c *Controller) processNextWorkItem() bool {
+	releaseQueueLength.Set(float64(c.releaseWorkqueue.Len()))
+
 	obj, shutdown := c.releaseWorkqueue.Get()
 	if shutdown {
 		return false
@@ -258,6 +260,7 @@ func (c *Controller) enqueueJob(obj interface{}) {
 		return
 	}
 	c.releaseWorkqueue.AddRateLimited(key)
+	releaseQueueLength.Set(float64(c.releaseWorkqueue.Len()))
 }
 
 // enqueueUpdateJob decides if there is a genuine resource update

integrations/helm/release/metrics.go

@@ -0,0 +1,38 @@
+package release
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/go-kit/kit/metrics/prometheus"
+	stdprometheus "github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	LabelAction      = "action"
+	LabelDryRun      = "dry_run"
+	LabelSuccess     = "success"
+	LabelNamespace   = "namespace"
+	LabelReleaseName = "release_name"
+)
+
+var (
+	durationBuckets = []float64{1, 5, 10, 30, 60, 120, 180, 300}
+	releaseDuration = prometheus.NewHistogramFrom(stdprometheus.HistogramOpts{
+		Namespace: "flux",
+		Subsystem: "helm_operator",
+		Name:      "release_duration_seconds",
+		Help:      "Release duration in seconds.",
+		Buckets:   durationBuckets,
+	}, []string{LabelAction, LabelDryRun, LabelSuccess, LabelNamespace, LabelReleaseName})
+)
+
+func ObserveRelease(start time.Time, action Action, dryRun, success bool, namespace, releaseName string) {
+	releaseDuration.With(
+		LabelAction, string(action),
+		LabelDryRun, fmt.Sprint(dryRun),
+		LabelSuccess, fmt.Sprint(success),
+		LabelNamespace, namespace,
+		LabelReleaseName, releaseName,
+	).Observe(time.Since(start).Seconds())
+}

integrations/helm/release/release.go

@@ -68,21 +68,6 @@ func New(logger log.Logger, helmClient *k8shelm.Client) *Release {
 	return r
 }
 
-// GetReleaseName either retrieves the release name from the Custom Resource or constructs a new one
-// in the form : $Namespace-$CustomResourceName
-func GetReleaseName(fhr flux_v1beta1.HelmRelease) string {
-	namespace := fhr.Namespace
-	if namespace == "" {
-		namespace = "default"
-	}
-	releaseName := fhr.Spec.ReleaseName
-	if releaseName == "" {
-		releaseName = fmt.Sprintf("%s-%s", namespace, fhr.Name)
-	}
-
-	return releaseName
-}
-
 // GetUpgradableRelease returns a release if the current state of it
 // allows an upgrade, a descriptive error if it is not allowed, or
 // nil if the release does not exist.
@@ -152,19 +137,30 @@ func (r *Release) canDelete(name string) (bool, error) {
 // TODO(michael): cloneDir is only relevant if installing from git;
 // either split this procedure into two varieties, or make it more
 // general and calculate the path to the chart in the caller.
-func (r *Release) Install(chartPath, releaseName string, fhr flux_v1beta1.HelmRelease, action Action, opts InstallOptions, kubeClient *kubernetes.Clientset) (*hapi_release.Release, error) {
+func (r *Release) Install(chartPath, releaseName string, fhr flux_v1beta1.HelmRelease, action Action, opts InstallOptions, kubeClient *kubernetes.Clientset) (release *hapi_release.Release, err error) {
+	defer func(start time.Time) {
+		ObserveRelease(
+			start,
+			action,
+			opts.DryRun,
+			err == nil,
+			fhr.Namespace,
+			fhr.ReleaseName(),
+		)
+	}(time.Now())
+
 	if chartPath == "" {
 		return nil, fmt.Errorf("empty path to chart supplied for resource %q", fhr.ResourceID().String())
 	}
-	_, err := os.Stat(chartPath)
+	_, err = os.Stat(chartPath)
 	switch {
 	case os.IsNotExist(err):
 		return nil, fmt.Errorf("no file or dir at path to chart: %s", chartPath)
 	case err != nil:
 		return nil, fmt.Errorf("error statting path given for chart %s: %s", chartPath, err.Error())
 	}
 
-	r.logger.Log("info", fmt.Sprintf("processing release %s (as %s)", GetReleaseName(fhr), releaseName),
+	r.logger.Log("info", fmt.Sprintf("processing release %s (as %s)", fhr.ReleaseName(), releaseName),
 		"action", fmt.Sprintf("%v", action),
 		"options", fmt.Sprintf("%+v", opts),
 		"timeout", fmt.Sprintf("%vs", fhr.GetTimeout()))

integrations/helm/status/status.go

@@ -24,7 +24,6 @@ import (
 	"github.com/weaveworks/flux/integrations/apis/flux.weave.works/v1beta1"
 	fluxclientset "github.com/weaveworks/flux/integrations/client/clientset/versioned"
 	v1beta1client "github.com/weaveworks/flux/integrations/client/clientset/versioned/typed/flux.weave.works/v1beta1"
-	"github.com/weaveworks/flux/integrations/helm/release"
 )
 
 const period = 10 * time.Second
@@ -79,7 +78,7 @@ bail:
 				break bail
 			}
 			for _, fhr := range fhrs.Items {
-				releaseName := release.GetReleaseName(fhr)
+				releaseName := fhr.ReleaseName()
 				// If we don't get the content, we don't care why
 				content, _ := a.helmClient.ReleaseContent(releaseName)
 				if content == nil {