Check metrics server availability during canary initialization

pkg/controller/scheduler.go

@@ -14,10 +14,6 @@ import (
  "github.com/weaveworks/flagger/pkg/router"
 )
 
-const (
- MetricsProviderServiceSuffix = ":service"
-)
-
 // scheduleCanaries synchronises the canary map with the jobs map,
 // for new canaries new jobs are created and started
 // for the removed canaries the jobs are stopped and deleted
@@ -119,6 +115,13 @@ func (c *Controller) advanceCanary(name string, namespace string) {
  return
  }
 
+ // check metric servers' availability
+ if !cd.SkipAnalysis() && (cd.Status.Phase == "" || cd.Status.Phase == flaggerv1.CanaryPhaseInitializing) {
+ if err := c.checkMetricProviderAvailability(cd); err != nil {
+ c.recordEventErrorf(cd, "Error checking metric providers: %v", err)
+ }
+ }
+
  // init mesh router
  meshRouter := c.routerFactory.MeshRouter(provider, labelSelector)
 

pkg/controller/scheduler_daemonset_fixture_test.go

@@ -79,7 +79,7 @@ func newDaemonSetFixture(c *flaggerv1.Canary) daemonSetFixture {
  rf := router.NewFactory(nil, kubeClient, flaggerClient, "annotationsPrefix", "", logger, flaggerClient)
 
  // init observer
- observerFactory, _ := observers.NewFactory("fake")
+ observerFactory, _ := observers.NewFactory(testMetricsServerURL)
 
  // init canary factory
  configTracker := &canary.ConfigTracker{
@@ -616,7 +616,7 @@ func newDaemonSetTestService() *corev1.Service {
 func newDaemonSetTestMetricTemplate() *flaggerv1.MetricTemplate {
  provider := flaggerv1.MetricTemplateProvider{
  Type: "prometheus",
- Address: "fake",
+ Address: testMetricsServerURL,
  SecretRef: &corev1.LocalObjectReference{
  Name: "podinfo-secret-env",
  },

pkg/controller/scheduler_deployment_fixture_test.go

@@ -107,7 +107,7 @@ func newDeploymentFixture(c *flaggerv1.Canary) fixture {
  rf := router.NewFactory(nil, kubeClient, flaggerClient, "annotationsPrefix", "", logger, flaggerClient)
 
  // init observer
- observerFactory, _ := observers.NewFactory("fake")
+ observerFactory, _ := observers.NewFactory(testMetricsServerURL)
 
  // init canary factory
  configTracker := &canary.ConfigTracker{
@@ -708,7 +708,7 @@ func newDeploymentTestHPA() *hpav2.HorizontalPodAutoscaler {
 func newDeploymentTestMetricTemplate() *flaggerv1.MetricTemplate {
  provider := flaggerv1.MetricTemplateProvider{
  Type: "prometheus",
- Address: "fake",
+ Address: testMetricsServerURL,
  SecretRef: &corev1.LocalObjectReference{
  Name: "podinfo-secret-env",
  },

pkg/controller/scheduler_deployment_test.go

@@ -84,11 +84,9 @@ func TestScheduler_DeploymentRollback(t *testing.T) {
 
  // run metric checks
  mocks.ctrl.advanceCanary("podinfo", "default")
- require.NoError(t, err)
 
  // finalise analysis
  mocks.ctrl.advanceCanary("podinfo", "default")
- require.NoError(t, err)
 
  // check status
  c, err = mocks.flaggerClient.FlaggerV1beta1().Canaries("default").Get(context.TODO(), "podinfo", metav1.GetOptions{})

pkg/controller/scheduler_metrics.go

@@ -3,6 +3,7 @@ package controller
 import (
  "context"
  "errors"
+ "fmt"
  "strings"
  "time"
 
@@ -13,6 +14,66 @@ import (
  "github.com/weaveworks/flagger/pkg/metrics/providers"
 )
 
+const (
+ MetricsProviderServiceSuffix = ":service"
+)
+
+// to be called during canary initialization
+func (c *Controller) checkMetricProviderAvailability(canary *flaggerv1.Canary) error {
+ for _, metric := range canary.GetAnalysis().Metrics {
+ if metric.Name == "request-success-rate" || metric.Name == "request-duration" {
+ observerFactory := c.observerFactory
+ if canary.Spec.MetricsServer != "" {
+ var err error
+ observerFactory, err = observers.NewFactory(canary.Spec.MetricsServer)
+ if err != nil {
+ return fmt.Errorf("error building Prometheus client for %s %v", canary.Spec.MetricsServer, err)
+ }
+ }
+ if ok, err := observerFactory.Client.IsOnline(); !ok || err != nil {
+ return fmt.Errorf("prometheus not avaiable: %v", err)
+ }
+ continue
+ }
+
+ if metric.TemplateRef != nil {
+ namespace := canary.Namespace
+ if metric.TemplateRef.Namespace != "" {
+ namespace = metric.TemplateRef.Namespace
+ }
+
+ template, err := c.flaggerInformers.MetricInformer.Lister().MetricTemplates(namespace).Get(metric.TemplateRef.Name)
+ if err != nil {
+ return fmt.Errorf("metric template %s.%s error: %v", metric.TemplateRef.Name, namespace, err)
+ }
+
+ var credentials map[string][]byte
+ if template.Spec.Provider.SecretRef != nil {
+ secret, err := c.kubeClient.CoreV1().Secrets(namespace).Get(context.TODO(), template.Spec.Provider.SecretRef.Name, metav1.GetOptions{})
+ if err != nil {
+ return fmt.Errorf("metric template %s.%s secret %s error: %v",
+ metric.TemplateRef.Name, namespace, template.Spec.Provider.SecretRef.Name, err)
+ }
+ credentials = secret.Data
+ }
+
+ factory := providers.Factory{}
+ provider, err := factory.Provider(metric.Interval, template.Spec.Provider, credentials)
+ if err != nil {
+ return fmt.Errorf("metric template %s.%s provider %s error: %v",
+ metric.TemplateRef.Name, namespace, template.Spec.Provider.Type, err)
+ }
+
+ if ok, err := provider.IsOnline(); !ok || err != nil {
+ return fmt.Errorf("%v in metric tempalte %s.%s not avaiable: %v", template.Spec.Provider.Type,
+ template.Name, template.Namespace, err)
+ }
+ }
+ }
+ c.recordEventInfof(canary, "all the metrics providers are available!")
+ return nil
+}
+
 func (c *Controller) runBuiltinMetricChecks(canary *flaggerv1.Canary) bool {
  // override the global provider if one is specified in the canary spec
  var metricsProvider string

pkg/controller/scheduler_metrics_test.go

@@ -0,0 +1,53 @@
+package controller
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+ "go.uber.org/zap"
+ "k8s.io/client-go/tools/record"
+
+ flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+ "github.com/weaveworks/flagger/pkg/metrics/observers"
+)
+
+func TestController_checkMetricProviderAvailability(t *testing.T) {
+ t.Run("builtin", func(t *testing.T) {
+ // ok
+ analysis := &flaggerv1.CanaryAnalysis{Metrics: []flaggerv1.CanaryMetric{{Name: "request-success-rate"}}}
+ canary := &flaggerv1.Canary{Spec: flaggerv1.CanarySpec{Analysis: analysis}}
+ obs, err := observers.NewFactory(testMetricsServerURL)
+ require.NoError(t, err)
+ ctrl := Controller{observerFactory: obs, logger: zap.S(), eventRecorder: &record.FakeRecorder{}}
+ require.NoError(t, ctrl.checkMetricProviderAvailability(canary))
+
+ // error
+ ctrl.observerFactory, err = observers.NewFactory("http://non-exist")
+ require.NoError(t, err)
+ require.Error(t, ctrl.checkMetricProviderAvailability(canary))
+
+ // ok
+ canary.Spec.MetricsServer = testMetricsServerURL
+ require.NoError(t, ctrl.checkMetricProviderAvailability(canary))
+ })
+
+ t.Run("templateRef", func(t *testing.T) {
+ ctrl := newDeploymentFixture(nil).ctrl
+
+ // error (not found)
+ analysis := &flaggerv1.CanaryAnalysis{Metrics: []flaggerv1.CanaryMetric{{
+ Name: "", TemplateRef: &flaggerv1.CrossNamespaceObjectReference{
+ Name: "non-exist", Namespace: "default",
+ },
+ }}}
+ canary := &flaggerv1.Canary{Spec: flaggerv1.CanarySpec{Analysis: analysis}}
+ require.Error(t, ctrl.checkMetricProviderAvailability(canary))
+
+ // ok
+ canary.Spec.Analysis.Metrics[0].TemplateRef = &flaggerv1.CrossNamespaceObjectReference{
+ Name: "envoy",
+ Namespace: "default",
+ }
+ require.NoError(t, ctrl.checkMetricProviderAvailability(canary))
+ })
+}

pkg/controller/scheduler_test.go

@@ -0,0 +1,54 @@
+package controller
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "testing"
+
+ flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+ clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
+
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var testMetricsServerURL string
+
+func TestMain(m *testing.M) {
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Query()["query"][0] == "vector(1)" {
+ // for IsOnline invoked during canary initialization
+ w.Write([]byte(`{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1545905245.458,"1"]}]}}`))
+ return
+ }
+ w.Write([]byte(`{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1545905245.458,"100"]}]}}`))
+ }))
+
+ testMetricsServerURL = ts.URL
+ defer ts.Close()
+ os.Exit(m.Run())
+}
+
+func assertPhase(flaggerClient clientset.Interface, canary string, phase flaggerv1.CanaryPhase) error {
+ c, err := flaggerClient.FlaggerV1beta1().Canaries("default").Get(context.TODO(), canary, metav1.GetOptions{})
+ if err != nil {
+ return err
+ }
+
+ if c.Status.Phase != phase {
+ return fmt.Errorf("got canary state %s wanted %s", c.Status.Phase, phase)
+ }
+
+ return nil
+}
+
+func alwaysReady() bool {
+ return true
+}
+
+func toFloatPtr(val int) *float64 {
+ v := float64(val)
+ return &v
+}

pkg/metrics/providers/prometheus.go

@@ -69,10 +69,6 @@ func NewPrometheusProvider(provider flaggerv1.MetricTemplateProvider, credential
 
 // RunQuery executes the promQL query and returns the the first result as float64
 func (p *PrometheusProvider) RunQuery(query string) (float64, error) {
- if p.url.String() == "fake" {
- return 100, nil
- }
-
  query = url.QueryEscape(p.trimQuery(query))
  u, err := url.Parse(fmt.Sprintf("./api/v1/query?query=%s", query))
  if err != nil {