Merge pull request #338 from weaveworks/appmesh-up

Implement App Mesh HTTP retry policy

artifacts/appmesh/canary.yaml

@@ -25,6 +25,11 @@ spec:
     portName: http
     # App Mesh reference
     meshName: global
+    # App Mesh retry policy (optional)
+    retries:
+      attempts: 3
+      perTryTimeout: 1s
+      retryOn: "gateway-error,client-error,stream-error"
   # define the canary analysis timing and KPIs
   canaryAnalysis:
     # schedule interval (default 60s)

docs/gitbook/faq.md

@@ -373,13 +373,16 @@ spec:
     # HTTP rewrite (optional)
     rewrite:
       uri: /
-    # Envoy timeout and retry policy (optional)
+    # Istio retry policy (optional)
+    retries:
+      attempts: 3
+      perTryTimeout: 1s
+      retryOn: "gateway-error,connect-failure,refused-stream"
+    # Add headers (optional)
     headers:
       request:
         add:
-          x-envoy-upstream-rq-timeout-ms: "15000"
-          x-envoy-max-retries: "10"
-          x-envoy-retry-on: "gateway-error,connect-failure,refused-stream"
+          x-some-header: "value"
     # cross-origin resource sharing policy (optional)
     corsPolicy:
       allowOrigin:
@@ -416,9 +419,7 @@ spec:
     - frontend
   http:
   - appendHeaders:
-      x-envoy-max-retries: "10"
-      x-envoy-retry-on: gateway-error,connect-failure,refused-stream
-      x-envoy-upstream-rq-timeout-ms: "15000"
+      x-some-header: "value"
     corsPolicy:
       allowHeaders:
       - x-some-header
@@ -439,6 +440,10 @@ spec:
     - destination:
         host: podinfo-canary
       weight: 0
+    retries:
+      attempts: 3
+      perTryTimeout: 1s
+      retryOn: "gateway-error,connect-failure,refused-stream"
 ```
 
 For each destination in the virtual service a rule is generated:

docs/gitbook/usage/appmesh-progressive-delivery.md

@@ -75,6 +75,11 @@ spec:
     # App Mesh egress (optional) 
     backends:
       - backend.test
+    # App Mesh retry policy (optional)
+    retries:
+      attempts: 3
+      perTryTimeout: 1s
+      retryOn: "gateway-error,client-error,stream-error"
   # define the canary analysis timing and KPIs
   canaryAnalysis:
     # schedule interval (default 60s)

docs/gitbook/usage/progressive-delivery.md

@@ -70,6 +70,11 @@ spec:
       tls:
         # use ISTIO_MUTUAL when mTLS is enabled
         mode: DISABLE
+    # Istio retry policy (optional)
+    retries:
+      attempts: 3
+      perTryTimeout: 1s
+      retryOn: "gateway-error,connect-failure,refused-stream"
   canaryAnalysis:
     # schedule interval (default 60s)
     interval: 1m

pkg/apis/appmesh/v1beta1/types.go

@@ -28,6 +28,7 @@ type Mesh struct {
 type MeshServiceDiscoveryType string
 
 const (
+	// Dns type is used when mesh is backed by a DNS namespace
 	Dns MeshServiceDiscoveryType = "Dns"
 )
 
@@ -104,8 +105,12 @@ type VirtualServiceSpec struct {
 
 // VirtualRouter is the spec for a VirtualRouter resource
 type VirtualRouter struct {
-	Name      string     `json:"name"`
-	Listeners []Listener `json:"listeners,omitempty"`
+	Name      string                  `json:"name"`
+	Listeners []VirtualRouterListener `json:"listeners,omitempty"`
+}
+
+type VirtualRouterListener struct {
+	PortMapping PortMapping `json:"portMapping"`
 }
 
 type Route struct {
@@ -114,21 +119,78 @@ type Route struct {
 	Http *HttpRoute `json:"http,omitempty"`
 	// +optional
 	Tcp *TcpRoute `json:"tcp,omitempty"`
+	// +optional
+	Priority *int64 `json:"priority,omitempty"`
 }
 
 type HttpRoute struct {
 	Match  HttpRouteMatch  `json:"match"`
 	Action HttpRouteAction `json:"action"`
+	// +optional
+	RetryPolicy *HttpRetryPolicy `json:"retryPolicy,omitempty"`
 }
 
 type HttpRouteMatch struct {
 	Prefix string `json:"prefix"`
+	// +optional
+	Method *string `json:"method,omitempty"`
+	// +optional
+	Headers []HttpRouteHeader `json:"headers,omitempty"`
+	// +optional
+	Scheme *string `json:"scheme,omitempty"`
+}
+
+type HttpRouteHeader struct {
+	Name string `json:"name"`
+	// +optional
+	Invert *bool `json:"invert,omitempty"`
+	// +optional
+	Match *HeaderMatchMethod `json:"match,omitempty"`
+}
+
+type HeaderMatchMethod struct {
+	// +optional
+	Exact *string `json:"exact,omitempty"`
+	// +optional
+	Prefix *string `json:"prefix,omitempty"`
+	// +optional
+	Range *MatchRange `json:"range,omitempty"`
+	// +optional
+	Regex *string `json:"regex,omitempty"`
+	// +optional
+	Suffix *string `json:"suffix,omitempty"`
+}
+
+type MatchRange struct {
+	// +optional
+	Start *int64 `json:"start,omitempty"`
+	// +optional
+	End *int64 `json:"end,omitempty"`
 }
 
 type HttpRouteAction struct {
 	WeightedTargets []WeightedTarget `json:"weightedTargets"`
 }
 
+type HttpRetryPolicy struct {
+	// +optional
+	PerRetryTimeoutMillis *int64 `json:"perRetryTimeoutMillis,omitempty"`
+	// +optional
+	MaxRetries *int64 `json:"maxRetries,omitempty"`
+	// +optional
+	HttpRetryPolicyEvents []HttpRetryPolicyEvent `json:"httpRetryEvents,omitempty"`
+	// +optional
+	TcpRetryPolicyEvents []TcpRetryPolicyEvent `json:"tcpRetryEvents,omitempty"`
+}
+
+type HttpRetryPolicyEvent string
+
+type TcpRetryPolicyEvent string
+
+const (
+	TcpRetryPolicyEventConnectionError TcpRetryPolicyEvent = "connection-error"
+)
+
 type TcpRoute struct {
 	Action TcpRouteAction `json:"action"`
 }
@@ -222,6 +284,8 @@ type VirtualNodeSpec struct {
 
 type Listener struct {
 	PortMapping PortMapping `json:"portMapping"`
+	// +optional
+	HealthCheck *HealthCheckPolicy `json:"healthCheck,omitempty"`
 }
 
 type PortMapping struct {
@@ -229,6 +293,28 @@ type PortMapping struct {
 	Protocol string `json:"protocol"`
 }
 
+type HealthCheckPolicy struct {
+	// +optional
+	HealthyThreshold *int64 `json:"healthyThreshold,omitempty"`
+	// +optional
+	IntervalMillis *int64 `json:"intervalMillis,omitempty"`
+	// +optional
+	Path *string `json:"path,omitempty"`
+	// +optional
+	Port *int64 `json:"port,omitempty"`
+	// +optional
+	Protocol *string `json:"protocol,omitempty"`
+	// +optional
+	TimeoutMillis *int64 `json:"timeoutMillis,omitempty"`
+	// +optional
+	UnhealthyThreshold *int64 `json:"unhealthyThreshold,omitempty"`
+}
+
+const (
+	PortProtocolHttp = "http"
+	PortProtocolTcp  = "tcp"
+)
+
 type ServiceDiscovery struct {
 	// +optional
 	CloudMap *CloudMapServiceDiscovery `json:"cloudMap,omitempty"`
@@ -237,7 +323,10 @@ type ServiceDiscovery struct {
 }
 
 type CloudMapServiceDiscovery struct {
-	CloudMapServiceName string `json:"cloudMapServiceName"`
+	ServiceName   string `json:"serviceName"`
+	NamespaceName string `json:"namespaceName"`
+	// +optional
+	Attributes map[string]string `json:"attributes,omitempty"`
 }
 
 type DnsServiceDiscovery struct {
@@ -272,13 +361,21 @@ type VirtualNodeStatus struct {
 	MeshArn *string `json:"meshArn,omitempty"`
 	// VirtualNodeArn is the AppMesh VirtualNode object's Amazon Resource Name
 	// +optional
-	VirtualNodeArn *string `json:"virtualNodeArn,omitempty"`
-	// CloudMapServiceArn is a CloudMap Service object's Amazon Resource Name
+	VirtualNodeArn *string                `json:"virtualNodeArn,omitempty"`
+	Conditions     []VirtualNodeCondition `json:"conditions"`
+	// CloudMapService is AWS CloudMap Service object's info
 	// +optional
-	CloudMapServiceArn *string `json:"cloudMapServiceArn,omitempty"`
+	CloudMapService *CloudMapServiceStatus `json:"cloudmapService,omitempty"`
+}
+
+// CloudMapServiceStatus is AWS CloudMap Service object's info
+type CloudMapServiceStatus struct {
+	// ServiceID is AWS CloudMap Service object's Id
 	// +optional
-	QueryParameters map[string]string      `json:"queryParameters,omitempty"`
-	Conditions      []VirtualNodeCondition `json:"conditions"`
+	ServiceID *string `json:"serviceId,omitempty"`
+	// NamespaceID is AWS CloudMap Service object's namespace Id
+	// +optional
+	NamespaceID *string `json:"namespaceId,omitempty"`
 }
 
 type VirtualNodeConditionType string

pkg/apis/appmesh/v1beta1/zz_generated.deepcopy.go

@@ -65,6 +65,13 @@ func (in *Backend) DeepCopy() *Backend {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CloudMapServiceDiscovery) DeepCopyInto(out *CloudMapServiceDiscovery) {
 	*out = *in
+	if in.Attributes != nil {
+		in, out := &in.Attributes, &out.Attributes
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 	return
 }
 
@@ -78,6 +85,32 @@ func (in *CloudMapServiceDiscovery) DeepCopy() *CloudMapServiceDiscovery {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CloudMapServiceStatus) DeepCopyInto(out *CloudMapServiceStatus) {
+	*out = *in
+	if in.ServiceID != nil {
+		in, out := &in.ServiceID, &out.ServiceID
+		*out = new(string)
+		**out = **in
+	}
+	if in.NamespaceID != nil {
+		in, out := &in.NamespaceID, &out.NamespaceID
+		*out = new(string)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CloudMapServiceStatus.
+func (in *CloudMapServiceStatus) DeepCopy() *CloudMapServiceStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(CloudMapServiceStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *DnsServiceDiscovery) DeepCopyInto(out *DnsServiceDiscovery) {
 	*out = *in
@@ -110,11 +143,144 @@ func (in *FileAccessLog) DeepCopy() *FileAccessLog {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *HeaderMatchMethod) DeepCopyInto(out *HeaderMatchMethod) {
+	*out = *in
+	if in.Exact != nil {
+		in, out := &in.Exact, &out.Exact
+		*out = new(string)
+		**out = **in
+	}
+	if in.Prefix != nil {
+		in, out := &in.Prefix, &out.Prefix
+		*out = new(string)
+		**out = **in
+	}
+	if in.Range != nil {
+		in, out := &in.Range, &out.Range
+		*out = new(MatchRange)
+		(*in).DeepCopyInto(*out)
+	}
+	if in.Regex != nil {
+		in, out := &in.Regex, &out.Regex
+		*out = new(string)
+		**out = **in
+	}
+	if in.Suffix != nil {
+		in, out := &in.Suffix, &out.Suffix
+		*out = new(string)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new HeaderMatchMethod.
+func (in *HeaderMatchMethod) DeepCopy() *HeaderMatchMethod {
+	if in == nil {
+		return nil
+	}
+	out := new(HeaderMatchMethod)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *HealthCheckPolicy) DeepCopyInto(out *HealthCheckPolicy) {
+	*out = *in
+	if in.HealthyThreshold != nil {
+		in, out := &in.HealthyThreshold, &out.HealthyThreshold
+		*out = new(int64)
+		**out = **in
+	}
+	if in.IntervalMillis != nil {
+		in, out := &in.IntervalMillis, &out.IntervalMillis
+		*out = new(int64)
+		**out = **in
+	}
+	if in.Path != nil {
+		in, out := &in.Path, &out.Path
+		*out = new(string)
+		**out = **in
+	}
+	if in.Port != nil {
+		in, out := &in.Port, &out.Port
+		*out = new(int64)
+		**out = **in
+	}
+	if in.Protocol != nil {
+		in, out := &in.Protocol, &out.Protocol
+		*out = new(string)
+		**out = **in
+	}
+	if in.TimeoutMillis != nil {
+		in, out := &in.TimeoutMillis, &out.TimeoutMillis
+		*out = new(int64)
+		**out = **in
+	}
+	if in.UnhealthyThreshold != nil {
+		in, out := &in.UnhealthyThreshold, &out.UnhealthyThreshold
+		*out = new(int64)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new HealthCheckPolicy.
+func (in *HealthCheckPolicy) DeepCopy() *HealthCheckPolicy {
+	if in == nil {
+		return nil
+	}
+	out := new(HealthCheckPolicy)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *HttpRetryPolicy) DeepCopyInto(out *HttpRetryPolicy) {
+	*out = *in
+	if in.PerRetryTimeoutMillis != nil {
+		in, out := &in.PerRetryTimeoutMillis, &out.PerRetryTimeoutMillis
+		*out = new(int64)
+		**out = **in
+	}
+	if in.MaxRetries != nil {
+		in, out := &in.MaxRetries, &out.MaxRetries
+		*out = new(int64)
+		**out = **in
+	}
+	if in.HttpRetryPolicyEvents != nil {
+		in, out := &in.HttpRetryPolicyEvents, &out.HttpRetryPolicyEvents
+		*out = make([]HttpRetryPolicyEvent, len(*in))
+		copy(*out, *in)
+	}
+	if in.TcpRetryPolicyEvents != nil {
+		in, out := &in.TcpRetryPolicyEvents, &out.TcpRetryPolicyEvents
+		*out = make([]TcpRetryPolicyEvent, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new HttpRetryPolicy.
+func (in *HttpRetryPolicy) DeepCopy() *HttpRetryPolicy {
+	if in == nil {
+		return nil
+	}
+	out := new(HttpRetryPolicy)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *HttpRoute) DeepCopyInto(out *HttpRoute) {
 	*out = *in
-	out.Match = in.Match
+	in.Match.DeepCopyInto(&out.Match)
 	in.Action.DeepCopyInto(&out.Action)
+	if in.RetryPolicy != nil {
+		in, out := &in.RetryPolicy, &out.RetryPolicy
+		*out = new(HttpRetryPolicy)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -149,9 +315,52 @@ func (in *HttpRouteAction) DeepCopy() *HttpRouteAction {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *HttpRouteHeader) DeepCopyInto(out *HttpRouteHeader) {
+	*out = *in
+	if in.Invert != nil {
+		in, out := &in.Invert, &out.Invert
+		*out = new(bool)
+		**out = **in
+	}
+	if in.Match != nil {
+		in, out := &in.Match, &out.Match
+		*out = new(HeaderMatchMethod)
+		(*in).DeepCopyInto(*out)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new HttpRouteHeader.
+func (in *HttpRouteHeader) DeepCopy() *HttpRouteHeader {
+	if in == nil {
+		return nil
+	}
+	out := new(HttpRouteHeader)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *HttpRouteMatch) DeepCopyInto(out *HttpRouteMatch) {
 	*out = *in
+	if in.Method != nil {
+		in, out := &in.Method, &out.Method
+		*out = new(string)
+		**out = **in
+	}
+	if in.Headers != nil {
+		in, out := &in.Headers, &out.Headers
+		*out = make([]HttpRouteHeader, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	if in.Scheme != nil {
+		in, out := &in.Scheme, &out.Scheme
+		*out = new(string)
+		**out = **in
+	}
 	return
 }
 
@@ -169,6 +378,11 @@ func (in *HttpRouteMatch) DeepCopy() *HttpRouteMatch {
 func (in *Listener) DeepCopyInto(out *Listener) {
 	*out = *in
 	out.PortMapping = in.PortMapping
+	if in.HealthCheck != nil {
+		in, out := &in.HealthCheck, &out.HealthCheck
+		*out = new(HealthCheckPolicy)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -203,6 +417,32 @@ func (in *Logging) DeepCopy() *Logging {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *MatchRange) DeepCopyInto(out *MatchRange) {
+	*out = *in
+	if in.Start != nil {
+		in, out := &in.Start, &out.Start
+		*out = new(int64)
+		**out = **in
+	}
+	if in.End != nil {
+		in, out := &in.End, &out.End
+		*out = new(int64)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new MatchRange.
+func (in *MatchRange) DeepCopy() *MatchRange {
+	if in == nil {
+		return nil
+	}
+	out := new(MatchRange)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Mesh) DeepCopyInto(out *Mesh) {
 	*out = *in
@@ -372,6 +612,11 @@ func (in *Route) DeepCopyInto(out *Route) {
 		*out = new(TcpRoute)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Priority != nil {
+		in, out := &in.Priority, &out.Priority
+		*out = new(int64)
+		**out = **in
+	}
 	return
 }
 
@@ -391,7 +636,7 @@ func (in *ServiceDiscovery) DeepCopyInto(out *ServiceDiscovery) {
 	if in.CloudMap != nil {
 		in, out := &in.CloudMap, &out.CloudMap
 		*out = new(CloudMapServiceDiscovery)
-		**out = **in
+		(*in).DeepCopyInto(*out)
 	}
 	if in.Dns != nil {
 		in, out := &in.Dns, &out.Dns
@@ -546,7 +791,9 @@ func (in *VirtualNodeSpec) DeepCopyInto(out *VirtualNodeSpec) {
 	if in.Listeners != nil {
 		in, out := &in.Listeners, &out.Listeners
 		*out = make([]Listener, len(*in))
-		copy(*out, *in)
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
 	}
 	if in.ServiceDiscovery != nil {
 		in, out := &in.ServiceDiscovery, &out.ServiceDiscovery
@@ -589,18 +836,6 @@ func (in *VirtualNodeStatus) DeepCopyInto(out *VirtualNodeStatus) {
 		*out = new(string)
 		**out = **in
 	}
-	if in.CloudMapServiceArn != nil {
-		in, out := &in.CloudMapServiceArn, &out.CloudMapServiceArn
-		*out = new(string)
-		**out = **in
-	}
-	if in.QueryParameters != nil {
-		in, out := &in.QueryParameters, &out.QueryParameters
-		*out = make(map[string]string, len(*in))
-		for key, val := range *in {
-			(*out)[key] = val
-		}
-	}
 	if in.Conditions != nil {
 		in, out := &in.Conditions, &out.Conditions
 		*out = make([]VirtualNodeCondition, len(*in))
@@ -608,6 +843,11 @@ func (in *VirtualNodeStatus) DeepCopyInto(out *VirtualNodeStatus) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.CloudMapService != nil {
+		in, out := &in.CloudMapService, &out.CloudMapService
+		*out = new(CloudMapServiceStatus)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -626,7 +866,7 @@ func (in *VirtualRouter) DeepCopyInto(out *VirtualRouter) {
 	*out = *in
 	if in.Listeners != nil {
 		in, out := &in.Listeners, &out.Listeners
-		*out = make([]Listener, len(*in))
+		*out = make([]VirtualRouterListener, len(*in))
 		copy(*out, *in)
 	}
 	return
@@ -642,6 +882,23 @@ func (in *VirtualRouter) DeepCopy() *VirtualRouter {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *VirtualRouterListener) DeepCopyInto(out *VirtualRouterListener) {
+	*out = *in
+	out.PortMapping = in.PortMapping
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new VirtualRouterListener.
+func (in *VirtualRouterListener) DeepCopy() *VirtualRouterListener {
+	if in == nil {
+		return nil
+	}
+	out := new(VirtualRouterListener)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *VirtualService) DeepCopyInto(out *VirtualService) {
 	*out = *in

pkg/router/appmesh.go

@@ -3,6 +3,7 @@ package router
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -181,7 +182,7 @@ func (ar *AppMeshRouter) reconcileVirtualService(canary *flaggerv1.Canary, name
 		MeshName: canary.Spec.Service.MeshName,
 		VirtualRouter: &appmeshv1.VirtualRouter{
 			Name: routerName,
-			Listeners: []appmeshv1.Listener{
+			Listeners: []appmeshv1.VirtualRouterListener{
 				{
 					PortMapping: appmeshv1.PortMapping{
 						Port:     int64(canary.Spec.Service.Port),
@@ -214,6 +215,33 @@ func (ar *AppMeshRouter) reconcileVirtualService(canary *flaggerv1.Canary, name
 		},
 	}
 
+	// add retry policy (default: one retry on gateway error with a 250ms timeout)
+	if canary.Spec.Service.Retries != nil {
+		timeout := int64(250)
+		if d, err := time.ParseDuration(canary.Spec.Service.Retries.PerTryTimeout); err == nil {
+			timeout = d.Milliseconds()
+		}
+
+		attempts := int64(1)
+		if canary.Spec.Service.Retries.Attempts > 0 {
+			attempts = int64(canary.Spec.Service.Retries.Attempts)
+		}
+
+		retryPolicy := &appmeshv1.HttpRetryPolicy{
+			PerRetryTimeoutMillis: int64p(timeout),
+			MaxRetries:            int64p(attempts),
+		}
+
+		events := []string{"gateway-error"}
+		if len(canary.Spec.Service.Retries.RetryOn) > 0 {
+			events = strings.Split(canary.Spec.Service.Retries.RetryOn, ",")
+		}
+		for _, value := range events {
+			retryPolicy.HttpRetryPolicyEvents = append(retryPolicy.HttpRetryPolicyEvents, appmeshv1.HttpRetryPolicyEvent(value))
+		}
+		vsSpec.Routes[0].Http.RetryPolicy = retryPolicy
+	}
+
 	virtualService, err := ar.appmeshClient.AppmeshV1beta1().VirtualServices(canary.Namespace).Get(name, metav1.GetOptions{})
 
 	// create virtual service
@@ -353,3 +381,7 @@ func getProtocol(canary *flaggerv1.Canary) string {
 	}
 	return "http"
 }
+
+func int64p(i int64) *int64 {
+	return &i
+}