mirror of
https://github.com/fluxcd/flagger.git
synced 2026-04-15 06:57:34 +00:00
pkg/metrics/providers: add AWS CloudWatch metrics provider
This commit is contained in:
mathetake
@@ -0,0 +1,112 @@
|
||||
package providers
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/awserr"
|
||||
"github.com/aws/aws-sdk-go/aws/session"
|
||||
"github.com/aws/aws-sdk-go/service/cloudwatch"
|
||||
|
||||
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
|
||||
)
|
||||
|
||||
const (
|
||||
cloudWatchMaxRetries = 3
|
||||
cloudWatchStartDeltaMultiplierOnMetricInterval = 10
|
||||
)
|
||||
|
||||
type CloudWatchProvider struct {
|
||||
client cloudWatchClient
|
||||
startDelta time.Duration
|
||||
}
|
||||
|
||||
// for the testing purpose
|
||||
type cloudWatchClient interface {
|
||||
GetMetricData(input *cloudwatch.GetMetricDataInput) (*cloudwatch.GetMetricDataOutput, error)
|
||||
}
|
||||
|
||||
// NewCloudWatchProvider takes a metricInterval, a provider spec and the credentials map, and
|
||||
// returns a cloudWatchProvider ready to execute queries against the AWS CloudWatch metrics
|
||||
func NewCloudWatchProvider(metricInterval string, provider flaggerv1.MetricTemplateProvider) (*CloudWatchProvider, error) {
|
||||
region := strings.TrimLeft(provider.Address, "monitoring.")
|
||||
region = strings.TrimRight(region, ".amazonaws.com")
|
||||
sess, err := session.NewSession(
|
||||
aws.NewConfig().
|
||||
WithRegion(region).
|
||||
WithMaxRetries(cloudWatchMaxRetries).
|
||||
WithEndpoint(provider.Address),
|
||||
)
|
||||
|
||||
md, err := time.ParseDuration(metricInterval)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error parsing metric interval: %s", err.Error())
|
||||
}
|
||||
|
||||
return &CloudWatchProvider{
|
||||
client: cloudwatch.New(sess),
|
||||
startDelta: cloudWatchStartDeltaMultiplierOnMetricInterval * md,
|
||||
}, err
|
||||
}
|
||||
|
||||
// RunQuery executes the aws cloud watch metrics query against GetMetricsData endpoint
|
||||
// and returns the the first result as float64
|
||||
func (p *CloudWatchProvider) RunQuery(query string) (float64, error) {
|
||||
var cq []*cloudwatch.MetricDataQuery
|
||||
if err := json.Unmarshal([]byte(query), &cq); err != nil {
|
||||
return 0, fmt.Errorf("error unmarshaling query: %s", err.Error())
|
||||
}
|
||||
|
||||
end := time.Now()
|
||||
start := end.Add(-p.startDelta)
|
||||
res, err := p.client.GetMetricData(&cloudwatch.GetMetricDataInput{
|
||||
EndTime: aws.Time(end),
|
||||
MaxDatapoints: aws.Int64(20),
|
||||
StartTime: aws.Time(start),
|
||||
MetricDataQueries: cq,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("error requesting cloudwatch: %s", err.Error())
|
||||
}
|
||||
|
||||
mr := res.MetricDataResults
|
||||
if len(mr) < 1 {
|
||||
return 0, fmt.Errorf("no values found in response: %s", res.String())
|
||||
}
|
||||
|
||||
vs := res.MetricDataResults[0].Values
|
||||
if len(vs) < 1 {
|
||||
return 0, fmt.Errorf("no values found in response: %s", res.String())
|
||||
}
|
||||
|
||||
return aws.Float64Value(vs[0]), nil
|
||||
}
|
||||
|
||||
// IsOnline calls GetMetricsData endpoint with the empty query
|
||||
// and returns an error if the returned status code is NOT http.StatusBadRequests.
|
||||
// For example, if the flagger does not have permission to perform `cloudwatch:GetMetricsData`,
|
||||
// the returned status code would be http.StatusForbidden
|
||||
func (p *CloudWatchProvider) IsOnline() (bool, error) {
|
||||
_, err := p.client.GetMetricData(&cloudwatch.GetMetricDataInput{
|
||||
EndTime: aws.Time(time.Time{}),
|
||||
MetricDataQueries: []*cloudwatch.MetricDataQuery{},
|
||||
StartTime: aws.Time(time.Time{}),
|
||||
})
|
||||
|
||||
if err == nil {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
ae, ok := err.(awserr.RequestFailure)
|
||||
if !ok {
|
||||
return false, fmt.Errorf("unexpected error: %v", err)
|
||||
} else if ae.StatusCode() != http.StatusBadRequest {
|
||||
return false, fmt.Errorf("unexpected status code: %v", ae)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
@@ -0,0 +1,177 @@
|
||||
package providers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/awserr"
|
||||
"github.com/aws/aws-sdk-go/service/cloudwatch"
|
||||
|
||||
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
|
||||
)
|
||||
|
||||
type cloudWatchClientMock struct {
|
||||
o *cloudwatch.GetMetricDataOutput
|
||||
err error
|
||||
}
|
||||
|
||||
func (c cloudWatchClientMock) GetMetricData(_ *cloudwatch.GetMetricDataInput) (*cloudwatch.GetMetricDataOutput, error) {
|
||||
return c.o, c.err
|
||||
}
|
||||
|
||||
func TestNewCloudWatchProvider(t *testing.T) {
|
||||
p, err := NewCloudWatchProvider(
|
||||
"5m",
|
||||
flaggerv1.MetricTemplateProvider{
|
||||
Address: "monitoring.ap-northeast-1.amazonaws.com",
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if exp := 5 * 60 * time.Second * cloudWatchStartDeltaMultiplierOnMetricInterval; p.startDelta != exp {
|
||||
t.Fatalf("expected %d but got %d", exp, p.startDelta)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCloudWatchProvider_IsOnline(t *testing.T) {
|
||||
t.Run("forbidden", func(t *testing.T) {
|
||||
p := CloudWatchProvider{client: cloudWatchClientMock{
|
||||
o: nil,
|
||||
err: awserr.NewRequestFailure(nil, http.StatusForbidden, "request-id"),
|
||||
}}
|
||||
|
||||
actual, err := p.IsOnline()
|
||||
if err == nil {
|
||||
t.Error("error expected")
|
||||
}
|
||||
if actual {
|
||||
t.Error("false expected")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("ok", func(t *testing.T) {
|
||||
// no error
|
||||
p := CloudWatchProvider{client: cloudWatchClientMock{}}
|
||||
actual, err := p.IsOnline()
|
||||
if err != nil {
|
||||
t.Errorf("no error expected but got %v", err)
|
||||
}
|
||||
if !actual {
|
||||
t.Error("true expected")
|
||||
}
|
||||
|
||||
// with error but bad request
|
||||
p = CloudWatchProvider{client: cloudWatchClientMock{
|
||||
err: awserr.NewRequestFailure(nil, http.StatusBadRequest, "request-id"),
|
||||
}}
|
||||
if err != nil {
|
||||
t.Errorf("no error expected but got %v", err)
|
||||
}
|
||||
if !actual {
|
||||
t.Error("true expected")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestCloudWatchProvider_RunQuery(t *testing.T) {
|
||||
// ref: https://aws.amazon.com/premiumsupport/knowledge-center/cloudwatch-getmetricdata-api/
|
||||
query := `
|
||||
[
|
||||
{
|
||||
"Id": "e1",
|
||||
"Expression": "m1 / m2",
|
||||
"Label": "ErrorRate"
|
||||
},
|
||||
{
|
||||
"Id": "m1",
|
||||
"MetricStat": {
|
||||
"Metric": {
|
||||
"Namespace": "MyApplication",
|
||||
"MetricName": "Errors",
|
||||
"Dimensions": [
|
||||
{
|
||||
"Name": "FunctionName",
|
||||
"Value": "MyFunc"
|
||||
}
|
||||
]
|
||||
},
|
||||
"Period": 300,
|
||||
"Stat": "Sum",
|
||||
"Unit": "Count"
|
||||
},
|
||||
"ReturnData": false
|
||||
},
|
||||
{
|
||||
"Id": "m2",
|
||||
"MetricStat": {
|
||||
"Metric": {
|
||||
"Namespace": "MyApplication",
|
||||
"MetricName": "Invocations",
|
||||
"Dimensions": [
|
||||
{
|
||||
"Name": "FunctionName",
|
||||
"Value": "MyFunc"
|
||||
}
|
||||
]
|
||||
},
|
||||
"Period": 300,
|
||||
"Stat": "Sum",
|
||||
"Unit": "Count"
|
||||
},
|
||||
"ReturnData": false
|
||||
}
|
||||
]`
|
||||
|
||||
t.Run("ok", func(t *testing.T) {
|
||||
var exp float64 = 100
|
||||
p := CloudWatchProvider{client: cloudWatchClientMock{
|
||||
o: &cloudwatch.GetMetricDataOutput{
|
||||
MetricDataResults: []*cloudwatch.MetricDataResult{
|
||||
{Values: []*float64{aws.Float64(exp)}},
|
||||
},
|
||||
},
|
||||
}}
|
||||
|
||||
actual, err := p.RunQuery(query)
|
||||
if err != nil {
|
||||
t.Fatalf("no error expected but got %v", err)
|
||||
}
|
||||
|
||||
if actual != exp {
|
||||
t.Errorf("expected %f but got %f", exp, actual)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("no values", func(t *testing.T) {
|
||||
assert := func(err error) {
|
||||
if err == nil {
|
||||
t.Fatal("error expected")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "no values") {
|
||||
t.Fatalf("no values expected in %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
p := CloudWatchProvider{client: cloudWatchClientMock{
|
||||
o: &cloudwatch.GetMetricDataOutput{
|
||||
MetricDataResults: []*cloudwatch.MetricDataResult{
|
||||
{Values: []*float64{}},
|
||||
},
|
||||
},
|
||||
}}
|
||||
|
||||
_, err := p.RunQuery(query)
|
||||
assert(err)
|
||||
|
||||
p = CloudWatchProvider{client: cloudWatchClientMock{
|
||||
o: &cloudwatch.GetMetricDataOutput{}}}
|
||||
|
||||
_, err = p.RunQuery(query)
|
||||
assert(err)
|
||||
})
|
||||
}
|
||||
@@ -11,12 +11,13 @@ func (factory Factory) Provider(
|
||||
provider flaggerv1.MetricTemplateProvider,
|
||||
credentials map[string][]byte,
|
||||
) (Interface, error) {
|
||||
|
||||
switch {
|
||||
case provider.Type == "prometheus":
|
||||
switch provider.Type {
|
||||
case "prometheus":
|
||||
return NewPrometheusProvider(provider, credentials)
|
||||
case provider.Type == "datadog":
|
||||
case "datadog":
|
||||
return NewDatadogProvider(metricInterval, provider, credentials)
|
||||
case "cloudwatch":
|
||||
return NewCloudWatchProvider(metricInterval, provider)
|
||||
default:
|
||||
return NewPrometheusProvider(provider, credentials)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user