feat: Web metric provider

metricproviders/metricproviders.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/argoproj/argo-rollouts/metricproviders/kayenta"
+	"github.com/argoproj/argo-rollouts/metricproviders/webmetric"
 
 	log "github.com/sirupsen/logrus"
 	"k8s.io/client-go/kubernetes"
@@ -50,6 +51,13 @@ func (f *ProviderFactory) NewProvider(logCtx log.Entry, metric v1alpha1.Metric)
 	} else if metric.Provider.Kayenta != nil {
 		c := kayenta.NewHttpClient()
 		return kayenta.NewKayentaProvider(logCtx, c), nil
+	} else if metric.Provider.WebMetric != nil {
+		c := webmetric.NewWebMetricHttpClient(metric)
+		p, err := webmetric.NewWebMetricJsonParser(metric)
+		if err != nil {
+			return nil, err
+		}
+		return webmetric.NewWebMetricProvider(logCtx, c, p), nil
 	}
 	return nil, fmt.Errorf("no valid provider in metric '%s'", metric.Name)
 }

metricproviders/webmetric/webmetric.go

@@ -0,0 +1,203 @@
+package webmetric
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+
+	"k8s.io/client-go/util/jsonpath"
+
+	"github.com/argoproj/argo-rollouts/utils/evaluate"
+	metricutil "github.com/argoproj/argo-rollouts/utils/metric"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	//ProviderType indicates the provider is prometheus
+	ProviderType = "WebMetric"
+)
+
+// Provider contains all the required components to run a WebMetric query
+// Implements the Provider Interface
+type Provider struct {
+	logCtx     log.Entry
+	client     *http.Client
+	jsonParser *jsonpath.JSONPath
+}
+
+// Type incidates provider is a WebMetric provider
+func (p *Provider) Type() string {
+	return ProviderType
+}
+
+func (p *Provider) Run(run *v1alpha1.AnalysisRun, metric v1alpha1.Metric) v1alpha1.Measurement {
+	var (
+		err error
+	)
+
+	startTime := metav1.Now()
+
+	// Measurement to pass back
+	measurement := v1alpha1.Measurement{
+		StartedAt: &startTime,
+	}
+
+	// Create request
+	request := &http.Request{
+		Method: "GET",
+	}
+
+	request.URL, err = url.Parse(metric.Provider.WebMetric.Url)
+	if err != nil {
+		return metricutil.MarkMeasurementError(measurement, err)
+	}
+
+	for _, header := range metric.Provider.WebMetric.Headers {
+		request.Header.Set(header.Key, header.Value)
+	}
+
+	// Send Request
+	response, err := p.client.Do(request)
+	if err != nil || response.StatusCode != http.StatusOK {
+		return metricutil.MarkMeasurementError(measurement, err)
+	}
+
+	value, status, err := p.parseResponse(metric, response)
+	if err != nil || response.StatusCode != http.StatusOK {
+		return metricutil.MarkMeasurementError(measurement, err)
+	}
+
+	measurement.Value = value
+	measurement.Phase = status
+	finishedTime := metav1.Now()
+	measurement.FinishedAt = &finishedTime
+
+	return measurement
+}
+
+func (p *Provider) parseResponse(metric v1alpha1.Metric, response *http.Response) (string, v1alpha1.AnalysisPhase, error) {
+	var data interface{}
+
+	bodyBytes, err := ioutil.ReadAll(response.Body)
+	if err != nil {
+		return "", v1alpha1.AnalysisPhaseError, fmt.Errorf("Received no bytes in response: %v", err)
+	}
+
+	err = json.Unmarshal(bodyBytes, &data)
+	if err != nil {
+		return "", v1alpha1.AnalysisPhaseError, fmt.Errorf("Could not parse JSON body: %v", err)
+	}
+
+	buf := new(bytes.Buffer)
+	err = p.jsonParser.Execute(buf, data)
+	if err != nil {
+		return "", v1alpha1.AnalysisPhaseError, fmt.Errorf("Could not find JSONPath in body: %s", err)
+	}
+	out := buf.String()
+	outAsFloat, err := strconv.ParseFloat(buf.String(), 64)
+	if err != nil {
+		return "", v1alpha1.AnalysisPhaseError, fmt.Errorf("Could not convert response to a number: %s", err)
+	}
+	status := p.evaluateResponse(metric, outAsFloat)
+	return out, status, nil
+}
+
+func (p *Provider) evaluateResponse(metric v1alpha1.Metric, result float64) v1alpha1.AnalysisPhase {
+	successCondition := false
+	failCondition := false
+	var err error
+
+	if metric.SuccessCondition != "" {
+		successCondition, err = evaluate.EvalCondition(result, metric.SuccessCondition)
+		if err != nil {
+			p.logCtx.Warning(err.Error())
+			return v1alpha1.AnalysisPhaseError
+		}
+	}
+	if metric.FailureCondition != "" {
+		failCondition, err = evaluate.EvalCondition(result, metric.FailureCondition)
+		if err != nil {
+			return v1alpha1.AnalysisPhaseError
+		}
+	}
+
+	switch {
+	case metric.SuccessCondition == "" && metric.FailureCondition == "":
+		//Always return success unless there is an error
+		return v1alpha1.AnalysisPhaseSuccessful
+	case metric.SuccessCondition != "" && metric.FailureCondition == "":
+		// Without a failure condition, a measurement is considered a failure if the measurement's success condition is not true
+		failCondition = !successCondition
+	case metric.SuccessCondition == "" && metric.FailureCondition != "":
+		// Without a success condition, a measurement is considered a successful if the measurement's failure condition is not true
+		successCondition = !failCondition
+	}
+
+	if failCondition {
+		return v1alpha1.AnalysisPhaseFailed
+	}
+
+	if !failCondition && !successCondition {
+		return v1alpha1.AnalysisPhaseInconclusive
+	}
+
+	// If we reach this code path, failCondition is false and successCondition is true
+	return v1alpha1.AnalysisPhaseSuccessful
+}
+
+// Resume should not be used the WebMetric provider since all the work should occur in the Run method
+func (p *Provider) Resume(run *v1alpha1.AnalysisRun, metric v1alpha1.Metric, measurement v1alpha1.Measurement) v1alpha1.Measurement {
+	p.logCtx.Warn("WebMetric provider should not execute the Resume method")
+	return measurement
+}
+
+// Terminate should not be used the WebMetric provider since all the work should occur in the Run method
+func (p *Provider) Terminate(run *v1alpha1.AnalysisRun, metric v1alpha1.Metric, measurement v1alpha1.Measurement) v1alpha1.Measurement {
+	p.logCtx.Warn("WebMetric provider should not execute the Terminate method")
+	return measurement
+}
+
+// GarbageCollect is a no-op for the WebMetric provider
+func (p *Provider) GarbageCollect(run *v1alpha1.AnalysisRun, metric v1alpha1.Metric, limit int) error {
+	return nil
+}
+
+func NewWebMetricHttpClient(metric v1alpha1.Metric) *http.Client {
+	var (
+		timeout time.Duration
+	)
+
+	if metric.Provider.WebMetric.Timeout <= 0 {
+		timeout = time.Duration(10) * time.Second
+	} else {
+		timeout = time.Duration(metric.Provider.WebMetric.Timeout) * time.Second
+	}
+	c := &http.Client{
+		Timeout: timeout,
+	}
+	return c
+}
+
+func NewWebMetricJsonParser(metric v1alpha1.Metric) (*jsonpath.JSONPath, error) {
+	jsonParser := jsonpath.New("metrics")
+
+	err := jsonParser.Parse(metric.Provider.WebMetric.JsonPath)
+
+	return jsonParser, err
+}
+
+func NewWebMetricProvider(logCtx log.Entry, client *http.Client, jsonParser *jsonpath.JSONPath) *Provider {
+	return &Provider{
+		logCtx:     logCtx,
+		client:     client,
+		jsonParser: jsonParser,
+	}
+}

metricproviders/webmetric/webmetric_test.go

@@ -0,0 +1,61 @@
+package webmetric
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRunSuccess(t *testing.T) {
+	input := `
+	{
+		"key": [
+			{
+				"key2": {
+					"value": 1
+				}
+			}
+		]
+	}`
+
+	server := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		rw.Header().Set("Content-Type", "application/json")
+		io.WriteString(rw, input)
+	}))
+	// Close the server when test finishes
+	defer server.Close()
+
+	e := log.Entry{}
+
+	metric := v1alpha1.Metric{
+		Name:             "foo",
+		SuccessCondition: "result > 0",
+		FailureCondition: "result <= 0",
+		Provider: v1alpha1.MetricProvider{
+			WebMetric: &v1alpha1.WebMetricMetric{
+				Url:      server.URL,
+				JsonPath: "{$.key[0].key2.value}",
+			},
+		},
+	}
+
+	jp, err := NewWebMetricJsonParser(metric)
+	assert.NoError(t, err)
+
+	p := NewWebMetricProvider(e, server.Client(), jp)
+
+	measurement := p.Run(newAnalysisRun(), metric)
+	assert.NotNil(t, measurement.StartedAt)
+	assert.Equal(t, "1", measurement.Value)
+	assert.NotNil(t, measurement.FinishedAt)
+	assert.Equal(t, v1alpha1.AnalysisPhaseSuccessful, measurement.Phase)
+}
+
+func newAnalysisRun() *v1alpha1.AnalysisRun {
+	return &v1alpha1.AnalysisRun{}
+}

pkg/apis/api-rules/violation_exceptions.list

@@ -1 +1,2 @@
+API rule violation: names_match,github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1,MetricProvider,WebMetric
 API rule violation: names_match,github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1,RolloutStatus,HPAReplicas

pkg/apis/rollouts/v1alpha1/analysis_types.go

@@ -97,6 +97,7 @@ type MetricProvider struct {
 	// Prometheus specifies the prometheus metric to query
 	Prometheus *PrometheusMetric `json:"prometheus,omitempty"`
 	Kayenta    *KayentaMetric    `json:"kayenta,omitempty"`
+	WebMetric  *WebMetricMetric  `json:"webmetric,omitempty"`
 	// Job specifies the job metric run
 	Job *JobMetric `json:"job,omitempty"`
 }
@@ -267,3 +268,16 @@ type ScopeDetail struct {
 	Start  string `json:"start"`
 	End    string `json:"end"`
 }
+
+// I don't like this repetition, but it fits the pattern...
+type WebMetricMetric struct {
+	Url      string            `json:"url"`
+	Headers  []WebMetricHeader `json:"headers,omitempty"`
+	Timeout  int               `json:"timeout,omitempty"`
+	JsonPath string            `json:"jsonPath"`
+}
+
+type WebMetricHeader struct {
+	Key   string `json:"key"`
+	Value string `json:"value"`
+}