github/flagger
1
0
Fork 0
mirror of https://github.com/fluxcd/flagger.git synced 2026-04-15 06:57:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move observer to metrics package

Browse Source
This commit is contained in:
stefanprodan
2019-03-30 11:45:39 +02:00
parent b945b37089
commit b2c12c1131
6 changed files with 22 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/controller/controller.go
+2 -6
View File
@@ -42,7 +42,7 @@ type Controller struct {
canaries *sync.Map
jobs map[string]CanaryJob
deployer CanaryDeployer
observer CanaryObserver
observer metrics.CanaryObserver
recorder metrics.CanaryRecorder
notifier *notifier.Slack
meshProvider string
@@ -81,10 +81,6 @@ func NewController(
},
}
observer := CanaryObserver{
metricsServer: metricServer,
}
recorder := metrics.NewCanaryRecorder(controllerAgentName, true)
recorder.SetInfo(version, meshProvider)
@@ -101,7 +97,7 @@ func NewController(
jobs: map[string]CanaryJob{},
flaggerWindow: flaggerWindow,
deployer: deployer,
observer: observer,
observer: metrics.NewObserver(metricServer),
recorder: recorder,
notifier: notifier,
meshProvider: meshProvider,
pkg/controller/controller_test.go
+2 -5
View File
@@ -35,7 +35,7 @@ type Mocks struct {
meshClient clientset.Interface
flaggerClient clientset.Interface
deployer CanaryDeployer
observer CanaryObserver
observer metrics.CanaryObserver
ctrl *Controller
logger *zap.SugaredLogger
router router.Interface
@@ -74,9 +74,6 @@ func SetupMocks(abtest bool) Mocks {
flaggerClient: flaggerClient,
},
}
observer := CanaryObserver{
metricsServer: "fake",
}
// init controller
flaggerInformerFactory := informers.NewSharedInformerFactory(flaggerClient, noResyncPeriodFunc())
@@ -94,7 +91,7 @@ func SetupMocks(abtest bool) Mocks {
canaries: new(sync.Map),
flaggerWindow: time.Second,
deployer: deployer,
observer: observer,
observer: metrics.NewObserver("fake"),
recorder: metrics.NewCanaryRecorder(controllerAgentName, false),
}
ctrl.flaggerSynced = alwaysReady
pkg/controller/observer.go
-257
View File
@@ -1,257 +0,0 @@
package controller
import (
"context"
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"net/url"
"strconv"
"strings"
"time"
)
// CanaryObserver is used to query the Istio Prometheus db
type CanaryObserver struct {
metricsServer string
}
type vectorQueryResponse struct {
Data struct {
Result []struct {
Metric struct {
Code string `json:"response_code"`
Name string `json:"destination_workload"`
}
Value []interface{} `json:"value"`
}
}
}
func (c *CanaryObserver) queryMetric(query string) (*vectorQueryResponse, error) {
promURL, err := url.Parse(c.metricsServer)
if err != nil {
return nil, err
}
u, err := url.Parse(fmt.Sprintf("./api/v1/query?query=%s", query))
if err != nil {
return nil, err
}
u = promURL.ResolveReference(u)
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(req.Context(), 5*time.Second)
defer cancel()
r, err := http.DefaultClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
}
defer r.Body.Close()
b, err := ioutil.ReadAll(r.Body)
if err != nil {
return nil, fmt.Errorf("error reading body: %s", err.Error())
}
if 400 <= r.StatusCode {
return nil, fmt.Errorf("error response: %s", string(b))
}
var values vectorQueryResponse
err = json.Unmarshal(b, &values)
if err != nil {
return nil, fmt.Errorf("error unmarshaling result: %s, '%s'", err.Error(), string(b))
}
return &values, nil
}
// GetScalar runs the promql query and returns the first value found
func (c *CanaryObserver) GetScalar(query string) (float64, error) {
if c.metricsServer == "fake" {
return 100, nil
}
query = strings.Replace(query, "\n", "", -1)
query = strings.Replace(query, " ", "", -1)
var value *float64
result, err := c.queryMetric(query)
if err != nil {
return 0, err
}
for _, v := range result.Data.Result {
metricValue := v.Value[1]
switch metricValue.(type) {
case string:
f, err := strconv.ParseFloat(metricValue.(string), 64)
if err != nil {
return 0, err
}
value = &f
}
}
if value == nil {
return 0, fmt.Errorf("no values found for query %s", query)
}
return *value, nil
}
func (c *CanaryObserver) GetEnvoySuccessRate(name string, namespace string, metric string, interval string) (float64, error) {
if c.metricsServer == "fake" {
return 100, nil
}
var rate *float64
querySt := url.QueryEscape(`sum(rate(` +
metric + `{kubernetes_namespace="` +
namespace + `",kubernetes_pod_name=~"` +
name + `-[0-9a-zA-Z]+(-[0-9a-zA-Z]+)",envoy_response_code!~"5.*"}[` +
interval + `])) / sum(rate(` +
metric + `{kubernetes_namespace="` +
namespace + `",kubernetes_pod_name=~"` +
name + `-[0-9a-zA-Z]+(-[0-9a-zA-Z]+)"}[` +
interval + `])) * 100 `)
result, err := c.queryMetric(querySt)
if err != nil {
return 0, err
}
for _, v := range result.Data.Result {
metricValue := v.Value[1]
switch metricValue.(type) {
case string:
f, err := strconv.ParseFloat(metricValue.(string), 64)
if err != nil {
return 0, err
}
rate = &f
}
}
if rate == nil {
return 0, fmt.Errorf("no values found for metric %s", metric)
}
return *rate, nil
}
// GetDeploymentCounter returns the requests success rate using istio_requests_total metric
func (c *CanaryObserver) GetDeploymentCounter(name string, namespace string, metric string, interval string) (float64, error) {
if c.metricsServer == "fake" {
return 100, nil
}
var rate *float64
querySt := url.QueryEscape(`sum(rate(` +
metric + `{reporter="destination",destination_workload_namespace=~"` +
namespace + `",destination_workload=~"` +
name + `",response_code!~"5.*"}[` +
interval + `])) / sum(rate(` +
metric + `{reporter="destination",destination_workload_namespace=~"` +
namespace + `",destination_workload=~"` +
name + `"}[` +
interval + `])) * 100 `)
result, err := c.queryMetric(querySt)
if err != nil {
return 0, err
}
for _, v := range result.Data.Result {
metricValue := v.Value[1]
switch metricValue.(type) {
case string:
f, err := strconv.ParseFloat(metricValue.(string), 64)
if err != nil {
return 0, err
}
rate = &f
}
}
if rate == nil {
return 0, fmt.Errorf("no values found for metric %s", metric)
}
return *rate, nil
}
// GetDeploymentHistogram returns the 99P requests delay using istio_request_duration_seconds_bucket metrics
func (c *CanaryObserver) GetDeploymentHistogram(name string, namespace string, metric string, interval string) (time.Duration, error) {
if c.metricsServer == "fake" {
return 1, nil
}
var rate *float64
querySt := url.QueryEscape(`histogram_quantile(0.99, sum(rate(` +
metric + `{reporter="destination",destination_workload=~"` +
name + `", destination_workload_namespace=~"` +
namespace + `"}[` +
interval + `])) by (le))`)
result, err := c.queryMetric(querySt)
if err != nil {
return 0, err
}
for _, v := range result.Data.Result {
metricValue := v.Value[1]
switch metricValue.(type) {
case string:
f, err := strconv.ParseFloat(metricValue.(string), 64)
if err != nil {
return 0, err
}
rate = &f
}
}
if rate == nil {
return 0, fmt.Errorf("no values found for metric %s", metric)
}
ms := time.Duration(int64(*rate*1000)) * time.Millisecond
return ms, nil
}
// CheckMetricsServer call Prometheus status endpoint and returns an error if
// the API is unreachable
func CheckMetricsServer(address string) (bool, error) {
promURL, err := url.Parse(address)
if err != nil {
return false, err
}
u, err := url.Parse("./api/v1/status/flags")
if err != nil {
return false, err
}
u = promURL.ResolveReference(u)
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return false, err
}
ctx, cancel := context.WithTimeout(req.Context(), 5*time.Second)
defer cancel()
r, err := http.DefaultClient.Do(req.WithContext(ctx))
if err != nil {
return false, err
}
defer r.Body.Close()
b, err := ioutil.ReadAll(r.Body)
if err != nil {
return false, fmt.Errorf("error reading body: %s", err.Error())
}
if 400 <= r.StatusCode {
return false, fmt.Errorf("error response: %s", string(b))
}
return true, nil
}
pkg/controller/observer_test.go
-84
View File
@@ -1,84 +0,0 @@
package controller
import (
"net/http"
"net/http/httptest"
"testing"
"time"
)
func TestCanaryObserver_GetDeploymentCounter(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1545905245.458,"100"]}]}}`
w.Write([]byte(json))
}))
defer ts.Close()
observer := CanaryObserver{
metricsServer: ts.URL,
}
val, err := observer.GetDeploymentCounter("podinfo", "default", "istio_requests_total", "1m")
if err != nil {
t.Fatal(err.Error())
}
if val != 100 {
t.Errorf("Got %v wanted %v", val, 100)
}
}
func TestCanaryObserver_GetDeploymentHistogram(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1545905245.596,"0.2"]}]}}`
w.Write([]byte(json))
}))
defer ts.Close()
observer := CanaryObserver{
metricsServer: ts.URL,
}
val, err := observer.GetDeploymentHistogram("podinfo", "default", "istio_request_duration_seconds_bucket", "1m")
if err != nil {
t.Fatal(err.Error())
}
if val != 200*time.Millisecond {
t.Errorf("Got %v wanted %v", val, 200*time.Millisecond)
}
}
func TestCheckMetricsServer(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
json := `{"status":"success","data":{"config.file":"/etc/prometheus/prometheus.yml"}}`
w.Write([]byte(json))
}))
defer ts.Close()
ok, err := CheckMetricsServer(ts.URL)
if err != nil {
t.Fatal(err.Error())
}
if !ok {
t.Errorf("Got %v wanted %v", ok, true)
}
}
func TestCheckMetricsServer_Offline(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusBadGateway)
}))
defer ts.Close()
ok, err := CheckMetricsServer(ts.URL)
if err == nil {
t.Errorf("Got no error wanted %v", http.StatusBadGateway)
}
if ok {
t.Errorf("Got %v wanted %v", ok, false)
}
}
pkg/controller/scheduler.go
+4 -4
View File
@@ -501,7 +501,7 @@ func (c *Controller) analyseCanary(r *flaggerv1.Canary) bool {
c.recordEventWarningf(r, "Halt advancement no values found for metric %s probably %s.%s is not receiving traffic",
metric.Name, r.Spec.TargetRef.Name, r.Namespace)
} else {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.metricsServer, err)
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
}
return false
}
@@ -519,7 +519,7 @@ func (c *Controller) analyseCanary(r *flaggerv1.Canary) bool {
c.recordEventWarningf(r, "Halt advancement no values found for metric %s probably %s.%s is not receiving traffic",
metric.Name, r.Spec.TargetRef.Name, r.Namespace)
} else {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.metricsServer, err)
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
}
return false
}
@@ -533,7 +533,7 @@ func (c *Controller) analyseCanary(r *flaggerv1.Canary) bool {
if metric.Name == "istio_request_duration_seconds_bucket" {
val, err := c.observer.GetDeploymentHistogram(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
if err != nil {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.metricsServer, err)
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
return false
}
t := time.Duration(metric.Threshold) * time.Millisecond
@@ -551,7 +551,7 @@ func (c *Controller) analyseCanary(r *flaggerv1.Canary) bool {
c.recordEventWarningf(r, "Halt advancement no values found for metric %s probably %s.%s is not receiving traffic",
metric.Name, r.Spec.TargetRef.Name, r.Namespace)
} else {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.metricsServer, err)
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
}
return false
}