Add CanaryDeployment kind

- bootstrap the deployments. services and Istio virtual service
- use google/go-cmp to detect changes in the deployment pod spec

pkg/apis/flagger/v1beta1/register.go

@@ -47,6 +47,8 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 	scheme.AddKnownTypes(SchemeGroupVersion,
 		&Canary{},
 		&CanaryList{},
+		&CanaryDeployment{},
+		&CanaryDeploymentList{},
 	)
 	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
 	return nil

pkg/apis/flagger/v1beta1/types.go

@@ -17,9 +17,12 @@ limitations under the License.
 package v1beta1
 
 import (
+	hpav1 "k8s.io/api/autoscaling/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const CanaryDeploymentKind  = "CanaryDeployment"
+
 // +genclient
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
@@ -41,6 +44,23 @@ type CanarySpec struct {
 	VirtualService VirtualService `json:"virtualService"`
 }
 
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// CanaryList is a list of Canary resources
+type CanaryList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata"`
+
+	Items []Canary `json:"items"`
+}
+
+// CanaryStatus is used for state persistence (read-only)
+type CanaryStatus struct {
+	State          string `json:"state"`
+	CanaryRevision string `json:"canaryRevision"`
+	FailedChecks   int    `json:"failedChecks"`
+}
+
 type Target struct {
 	Name string `json:"name"`
 	Host string `json:"host"`
@@ -63,19 +83,49 @@ type Metric struct {
 	Threshold int    `json:"threshold"`
 }
 
-// CanaryStatus is the status for a Canary resource
-type CanaryStatus struct {
-	State          string `json:"state"`
-	CanaryRevision string `json:"canaryRevision"`
-	FailedChecks   int    `json:"failedChecks"`
+// +genclient
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// Canary is a specification for a Canary resource
+type CanaryDeployment struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   CanaryDeploymentSpec   `json:"spec"`
+	Status CanaryDeploymentStatus `json:"status"`
+}
+
+// CanarySpec is the spec for a Canary resource
+type CanaryDeploymentSpec struct {
+	// reference to target resource
+	TargetRef hpav1.CrossVersionObjectReference `json:"targetRef"`
+
+	// virtual service spec
+	Service CanaryDeploymentService `json:"service"`
+
+	// metrics and thresholds
+	CanaryAnalysis CanaryAnalysis `json:"canaryAnalysis"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // CanaryList is a list of Canary resources
-type CanaryList struct {
+type CanaryDeploymentList struct {
 	metav1.TypeMeta `json:",inline"`
 	metav1.ListMeta `json:"metadata"`
 
-	Items []Canary `json:"items"`
+	Items []CanaryDeployment `json:"items"`
+}
+
+// CanaryStatus is used for state persistence (read-only)
+type CanaryDeploymentStatus struct {
+	State          string `json:"state"`
+	CanaryRevision string `json:"canaryRevision"`
+	FailedChecks   int    `json:"failedChecks"`
+}
+
+type CanaryDeploymentService struct {
+	Port     int32      `json:"port"`
+	Gateways []string `json:"gateways"`
+	Hosts    []string `json:"hosts"`
 }

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

@@ -73,6 +73,128 @@ func (in *CanaryAnalysis) DeepCopy() *CanaryAnalysis {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CanaryDeployment) DeepCopyInto(out *CanaryDeployment) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	in.Spec.DeepCopyInto(&out.Spec)
+	out.Status = in.Status
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CanaryDeployment.
+func (in *CanaryDeployment) DeepCopy() *CanaryDeployment {
+	if in == nil {
+		return nil
+	}
+	out := new(CanaryDeployment)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *CanaryDeployment) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CanaryDeploymentList) DeepCopyInto(out *CanaryDeploymentList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	out.ListMeta = in.ListMeta
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]CanaryDeployment, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CanaryDeploymentList.
+func (in *CanaryDeploymentList) DeepCopy() *CanaryDeploymentList {
+	if in == nil {
+		return nil
+	}
+	out := new(CanaryDeploymentList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *CanaryDeploymentList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CanaryDeploymentService) DeepCopyInto(out *CanaryDeploymentService) {
+	*out = *in
+	if in.Gateways != nil {
+		in, out := &in.Gateways, &out.Gateways
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.Hosts != nil {
+		in, out := &in.Hosts, &out.Hosts
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CanaryDeploymentService.
+func (in *CanaryDeploymentService) DeepCopy() *CanaryDeploymentService {
+	if in == nil {
+		return nil
+	}
+	out := new(CanaryDeploymentService)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CanaryDeploymentSpec) DeepCopyInto(out *CanaryDeploymentSpec) {
+	*out = *in
+	out.TargetRef = in.TargetRef
+	in.Service.DeepCopyInto(&out.Service)
+	in.CanaryAnalysis.DeepCopyInto(&out.CanaryAnalysis)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CanaryDeploymentSpec.
+func (in *CanaryDeploymentSpec) DeepCopy() *CanaryDeploymentSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(CanaryDeploymentSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CanaryDeploymentStatus) DeepCopyInto(out *CanaryDeploymentStatus) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CanaryDeploymentStatus.
+func (in *CanaryDeploymentStatus) DeepCopy() *CanaryDeploymentStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(CanaryDeploymentStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CanaryList) DeepCopyInto(out *CanaryList) {
 	*out = *in

pkg/client/clientset/versioned/typed/flagger/v1beta1/canarydeployment.go

@@ -0,0 +1,174 @@
+/*
+Copyright The Flagger Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	v1beta1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1beta1"
+	scheme "github.com/stefanprodan/flagger/pkg/client/clientset/versioned/scheme"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	rest "k8s.io/client-go/rest"
+)
+
+// CanaryDeploymentsGetter has a method to return a CanaryDeploymentInterface.
+// A group's client should implement this interface.
+type CanaryDeploymentsGetter interface {
+	CanaryDeployments(namespace string) CanaryDeploymentInterface
+}
+
+// CanaryDeploymentInterface has methods to work with CanaryDeployment resources.
+type CanaryDeploymentInterface interface {
+	Create(*v1beta1.CanaryDeployment) (*v1beta1.CanaryDeployment, error)
+	Update(*v1beta1.CanaryDeployment) (*v1beta1.CanaryDeployment, error)
+	UpdateStatus(*v1beta1.CanaryDeployment) (*v1beta1.CanaryDeployment, error)
+	Delete(name string, options *v1.DeleteOptions) error
+	DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error
+	Get(name string, options v1.GetOptions) (*v1beta1.CanaryDeployment, error)
+	List(opts v1.ListOptions) (*v1beta1.CanaryDeploymentList, error)
+	Watch(opts v1.ListOptions) (watch.Interface, error)
+	Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1beta1.CanaryDeployment, err error)
+	CanaryDeploymentExpansion
+}
+
+// canaryDeployments implements CanaryDeploymentInterface
+type canaryDeployments struct {
+	client rest.Interface
+	ns     string
+}
+
+// newCanaryDeployments returns a CanaryDeployments
+func newCanaryDeployments(c *FlaggerV1beta1Client, namespace string) *canaryDeployments {
+	return &canaryDeployments{
+		client: c.RESTClient(),
+		ns:     namespace,
+	}
+}
+
+// Get takes name of the canaryDeployment, and returns the corresponding canaryDeployment object, and an error if there is any.
+func (c *canaryDeployments) Get(name string, options v1.GetOptions) (result *v1beta1.CanaryDeployment, err error) {
+	result = &v1beta1.CanaryDeployment{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		Name(name).
+		VersionedParams(&options, scheme.ParameterCodec).
+		Do().
+		Into(result)
+	return
+}
+
+// List takes label and field selectors, and returns the list of CanaryDeployments that match those selectors.
+func (c *canaryDeployments) List(opts v1.ListOptions) (result *v1beta1.CanaryDeploymentList, err error) {
+	result = &v1beta1.CanaryDeploymentList{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Do().
+		Into(result)
+	return
+}
+
+// Watch returns a watch.Interface that watches the requested canaryDeployments.
+func (c *canaryDeployments) Watch(opts v1.ListOptions) (watch.Interface, error) {
+	opts.Watch = true
+	return c.client.Get().
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Watch()
+}
+
+// Create takes the representation of a canaryDeployment and creates it.  Returns the server's representation of the canaryDeployment, and an error, if there is any.
+func (c *canaryDeployments) Create(canaryDeployment *v1beta1.CanaryDeployment) (result *v1beta1.CanaryDeployment, err error) {
+	result = &v1beta1.CanaryDeployment{}
+	err = c.client.Post().
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		Body(canaryDeployment).
+		Do().
+		Into(result)
+	return
+}
+
+// Update takes the representation of a canaryDeployment and updates it. Returns the server's representation of the canaryDeployment, and an error, if there is any.
+func (c *canaryDeployments) Update(canaryDeployment *v1beta1.CanaryDeployment) (result *v1beta1.CanaryDeployment, err error) {
+	result = &v1beta1.CanaryDeployment{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		Name(canaryDeployment.Name).
+		Body(canaryDeployment).
+		Do().
+		Into(result)
+	return
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+
+func (c *canaryDeployments) UpdateStatus(canaryDeployment *v1beta1.CanaryDeployment) (result *v1beta1.CanaryDeployment, err error) {
+	result = &v1beta1.CanaryDeployment{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		Name(canaryDeployment.Name).
+		SubResource("status").
+		Body(canaryDeployment).
+		Do().
+		Into(result)
+	return
+}
+
+// Delete takes name of the canaryDeployment and deletes it. Returns an error if one occurs.
+func (c *canaryDeployments) Delete(name string, options *v1.DeleteOptions) error {
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		Name(name).
+		Body(options).
+		Do().
+		Error()
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *canaryDeployments) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		VersionedParams(&listOptions, scheme.ParameterCodec).
+		Body(options).
+		Do().
+		Error()
+}
+
+// Patch applies the patch and returns the patched canaryDeployment.
+func (c *canaryDeployments) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1beta1.CanaryDeployment, err error) {
+	result = &v1beta1.CanaryDeployment{}
+	err = c.client.Patch(pt).
+		Namespace(c.ns).
+		Resource("canarydeployments").
+		SubResource(subresources...).
+		Name(name).
+		Body(data).
+		Do().
+		Into(result)
+	return
+}

pkg/client/clientset/versioned/typed/flagger/v1beta1/fake/fake_canarydeployment.go

@@ -0,0 +1,140 @@
+/*
+Copyright The Flagger Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	v1beta1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1beta1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+)
+
+// FakeCanaryDeployments implements CanaryDeploymentInterface
+type FakeCanaryDeployments struct {
+	Fake *FakeFlaggerV1beta1
+	ns   string
+}
+
+var canarydeploymentsResource = schema.GroupVersionResource{Group: "flagger.app", Version: "v1beta1", Resource: "canarydeployments"}
+
+var canarydeploymentsKind = schema.GroupVersionKind{Group: "flagger.app", Version: "v1beta1", Kind: "CanaryDeployment"}
+
+// Get takes name of the canaryDeployment, and returns the corresponding canaryDeployment object, and an error if there is any.
+func (c *FakeCanaryDeployments) Get(name string, options v1.GetOptions) (result *v1beta1.CanaryDeployment, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(canarydeploymentsResource, c.ns, name), &v1beta1.CanaryDeployment{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.CanaryDeployment), err
+}
+
+// List takes label and field selectors, and returns the list of CanaryDeployments that match those selectors.
+func (c *FakeCanaryDeployments) List(opts v1.ListOptions) (result *v1beta1.CanaryDeploymentList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(canarydeploymentsResource, canarydeploymentsKind, c.ns, opts), &v1beta1.CanaryDeploymentList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1beta1.CanaryDeploymentList{ListMeta: obj.(*v1beta1.CanaryDeploymentList).ListMeta}
+	for _, item := range obj.(*v1beta1.CanaryDeploymentList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested canaryDeployments.
+func (c *FakeCanaryDeployments) Watch(opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(canarydeploymentsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a canaryDeployment and creates it.  Returns the server's representation of the canaryDeployment, and an error, if there is any.
+func (c *FakeCanaryDeployments) Create(canaryDeployment *v1beta1.CanaryDeployment) (result *v1beta1.CanaryDeployment, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(canarydeploymentsResource, c.ns, canaryDeployment), &v1beta1.CanaryDeployment{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.CanaryDeployment), err
+}
+
+// Update takes the representation of a canaryDeployment and updates it. Returns the server's representation of the canaryDeployment, and an error, if there is any.
+func (c *FakeCanaryDeployments) Update(canaryDeployment *v1beta1.CanaryDeployment) (result *v1beta1.CanaryDeployment, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(canarydeploymentsResource, c.ns, canaryDeployment), &v1beta1.CanaryDeployment{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.CanaryDeployment), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeCanaryDeployments) UpdateStatus(canaryDeployment *v1beta1.CanaryDeployment) (*v1beta1.CanaryDeployment, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(canarydeploymentsResource, "status", c.ns, canaryDeployment), &v1beta1.CanaryDeployment{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.CanaryDeployment), err
+}
+
+// Delete takes name of the canaryDeployment and deletes it. Returns an error if one occurs.
+func (c *FakeCanaryDeployments) Delete(name string, options *v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(canarydeploymentsResource, c.ns, name), &v1beta1.CanaryDeployment{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeCanaryDeployments) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(canarydeploymentsResource, c.ns, listOptions)
+
+	_, err := c.Fake.Invokes(action, &v1beta1.CanaryDeploymentList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched canaryDeployment.
+func (c *FakeCanaryDeployments) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1beta1.CanaryDeployment, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(canarydeploymentsResource, c.ns, name, data, subresources...), &v1beta1.CanaryDeployment{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.CanaryDeployment), err
+}

pkg/client/clientset/versioned/typed/flagger/v1beta1/fake/fake_flagger_client.go

@@ -32,6 +32,10 @@ func (c *FakeFlaggerV1beta1) Canaries(namespace string) v1beta1.CanaryInterface
 	return &FakeCanaries{c, namespace}
 }
 
+func (c *FakeFlaggerV1beta1) CanaryDeployments(namespace string) v1beta1.CanaryDeploymentInterface {
+	return &FakeCanaryDeployments{c, namespace}
+}
+
 // RESTClient returns a RESTClient that is used to communicate
 // with API server by this client implementation.
 func (c *FakeFlaggerV1beta1) RESTClient() rest.Interface {

pkg/client/clientset/versioned/typed/flagger/v1beta1/flagger_client.go

@@ -28,6 +28,7 @@ import (
 type FlaggerV1beta1Interface interface {
 	RESTClient() rest.Interface
 	CanariesGetter
+	CanaryDeploymentsGetter
 }
 
 // FlaggerV1beta1Client is used to interact with features provided by the flagger.app group.
@@ -39,6 +40,10 @@ func (c *FlaggerV1beta1Client) Canaries(namespace string) CanaryInterface {
 	return newCanaries(c, namespace)
 }
 
+func (c *FlaggerV1beta1Client) CanaryDeployments(namespace string) CanaryDeploymentInterface {
+	return newCanaryDeployments(c, namespace)
+}
+
 // NewForConfig creates a new FlaggerV1beta1Client for the given config.
 func NewForConfig(c *rest.Config) (*FlaggerV1beta1Client, error) {
 	config := *c

pkg/client/clientset/versioned/typed/flagger/v1beta1/generated_expansion.go

@@ -19,3 +19,5 @@ limitations under the License.
 package v1beta1
 
 type CanaryExpansion interface{}
+
+type CanaryDeploymentExpansion interface{}

pkg/client/informers/externalversions/flagger/v1beta1/canarydeployment.go

@@ -0,0 +1,89 @@
+/*
+Copyright The Flagger Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	time "time"
+
+	flaggerv1beta1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1beta1"
+	versioned "github.com/stefanprodan/flagger/pkg/client/clientset/versioned"
+	internalinterfaces "github.com/stefanprodan/flagger/pkg/client/informers/externalversions/internalinterfaces"
+	v1beta1 "github.com/stefanprodan/flagger/pkg/client/listers/flagger/v1beta1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+)
+
+// CanaryDeploymentInformer provides access to a shared informer and lister for
+// CanaryDeployments.
+type CanaryDeploymentInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1beta1.CanaryDeploymentLister
+}
+
+type canaryDeploymentInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewCanaryDeploymentInformer constructs a new informer for CanaryDeployment type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewCanaryDeploymentInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredCanaryDeploymentInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredCanaryDeploymentInformer constructs a new informer for CanaryDeployment type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredCanaryDeploymentInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.FlaggerV1beta1().CanaryDeployments(namespace).List(options)
+			},
+			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.FlaggerV1beta1().CanaryDeployments(namespace).Watch(options)
+			},
+		},
+		&flaggerv1beta1.CanaryDeployment{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *canaryDeploymentInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredCanaryDeploymentInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *canaryDeploymentInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&flaggerv1beta1.CanaryDeployment{}, f.defaultInformer)
+}
+
+func (f *canaryDeploymentInformer) Lister() v1beta1.CanaryDeploymentLister {
+	return v1beta1.NewCanaryDeploymentLister(f.Informer().GetIndexer())
+}

pkg/client/informers/externalversions/flagger/v1beta1/interface.go

@@ -26,6 +26,8 @@ import (
 type Interface interface {
 	// Canaries returns a CanaryInformer.
 	Canaries() CanaryInformer
+	// CanaryDeployments returns a CanaryDeploymentInformer.
+	CanaryDeployments() CanaryDeploymentInformer
 }
 
 type version struct {
@@ -43,3 +45,8 @@ func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakList
 func (v *version) Canaries() CanaryInformer {
 	return &canaryInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
 }
+
+// CanaryDeployments returns a CanaryDeploymentInformer.
+func (v *version) CanaryDeployments() CanaryDeploymentInformer {
+	return &canaryDeploymentInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}

pkg/client/informers/externalversions/generic.go

@@ -55,6 +55,8 @@ func (f *sharedInformerFactory) ForResource(resource schema.GroupVersionResource
 	// Group=flagger.app, Version=v1beta1
 	case v1beta1.SchemeGroupVersion.WithResource("canaries"):
 		return &genericInformer{resource: resource.GroupResource(), informer: f.Flagger().V1beta1().Canaries().Informer()}, nil
+	case v1beta1.SchemeGroupVersion.WithResource("canarydeployments"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Flagger().V1beta1().CanaryDeployments().Informer()}, nil
 
 	}
 

pkg/client/listers/flagger/v1beta1/canarydeployment.go

@@ -0,0 +1,94 @@
+/*
+Copyright The Flagger Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	v1beta1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1beta1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+)
+
+// CanaryDeploymentLister helps list CanaryDeployments.
+type CanaryDeploymentLister interface {
+	// List lists all CanaryDeployments in the indexer.
+	List(selector labels.Selector) (ret []*v1beta1.CanaryDeployment, err error)
+	// CanaryDeployments returns an object that can list and get CanaryDeployments.
+	CanaryDeployments(namespace string) CanaryDeploymentNamespaceLister
+	CanaryDeploymentListerExpansion
+}
+
+// canaryDeploymentLister implements the CanaryDeploymentLister interface.
+type canaryDeploymentLister struct {
+	indexer cache.Indexer
+}
+
+// NewCanaryDeploymentLister returns a new CanaryDeploymentLister.
+func NewCanaryDeploymentLister(indexer cache.Indexer) CanaryDeploymentLister {
+	return &canaryDeploymentLister{indexer: indexer}
+}
+
+// List lists all CanaryDeployments in the indexer.
+func (s *canaryDeploymentLister) List(selector labels.Selector) (ret []*v1beta1.CanaryDeployment, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1beta1.CanaryDeployment))
+	})
+	return ret, err
+}
+
+// CanaryDeployments returns an object that can list and get CanaryDeployments.
+func (s *canaryDeploymentLister) CanaryDeployments(namespace string) CanaryDeploymentNamespaceLister {
+	return canaryDeploymentNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// CanaryDeploymentNamespaceLister helps list and get CanaryDeployments.
+type CanaryDeploymentNamespaceLister interface {
+	// List lists all CanaryDeployments in the indexer for a given namespace.
+	List(selector labels.Selector) (ret []*v1beta1.CanaryDeployment, err error)
+	// Get retrieves the CanaryDeployment from the indexer for a given namespace and name.
+	Get(name string) (*v1beta1.CanaryDeployment, error)
+	CanaryDeploymentNamespaceListerExpansion
+}
+
+// canaryDeploymentNamespaceLister implements the CanaryDeploymentNamespaceLister
+// interface.
+type canaryDeploymentNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all CanaryDeployments in the indexer for a given namespace.
+func (s canaryDeploymentNamespaceLister) List(selector labels.Selector) (ret []*v1beta1.CanaryDeployment, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1beta1.CanaryDeployment))
+	})
+	return ret, err
+}
+
+// Get retrieves the CanaryDeployment from the indexer for a given namespace and name.
+func (s canaryDeploymentNamespaceLister) Get(name string) (*v1beta1.CanaryDeployment, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1beta1.Resource("canarydeployment"), name)
+	}
+	return obj.(*v1beta1.CanaryDeployment), nil
+}

pkg/client/listers/flagger/v1beta1/expansion_generated.go

@@ -25,3 +25,11 @@ type CanaryListerExpansion interface{}
 // CanaryNamespaceListerExpansion allows custom methods to be added to
 // CanaryNamespaceLister.
 type CanaryNamespaceListerExpansion interface{}
+
+// CanaryDeploymentListerExpansion allows custom methods to be added to
+// CanaryDeploymentLister.
+type CanaryDeploymentListerExpansion interface{}
+
+// CanaryDeploymentNamespaceListerExpansion allows custom methods to be added to
+// CanaryDeploymentNamespaceLister.
+type CanaryDeploymentNamespaceListerExpansion interface{}

pkg/operator/deployer.go

@@ -0,0 +1,246 @@
+package operator
+
+import (
+	"fmt"
+
+	istiov1alpha3 "github.com/knative/pkg/apis/istio/v1alpha3"
+	flaggerv1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1beta1"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+func (c *Controller) bootstrapDeployment(cd *flaggerv1.CanaryDeployment) error {
+
+	canaryName := cd.Spec.TargetRef.Name
+	primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
+
+	canaryDep, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(canaryName, metav1.GetOptions{})
+	if err != nil {
+		if errors.IsNotFound(err) {
+			return fmt.Errorf("deployment %s.%s not found, retrying in %v",
+				canaryName, cd.Namespace, c.rolloutWindow)
+		} else {
+			return err
+		}
+	}
+
+	primaryDep, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(primaryName, metav1.GetOptions{})
+	if errors.IsNotFound(err) {
+		primaryDep = &appsv1.Deployment{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        primaryName,
+				Annotations: canaryDep.Annotations,
+				Namespace:   cd.Namespace,
+				OwnerReferences: []metav1.OwnerReference{
+					*metav1.NewControllerRef(cd, schema.GroupVersionKind{
+						Group:   flaggerv1.SchemeGroupVersion.Group,
+						Version: flaggerv1.SchemeGroupVersion.Version,
+						Kind:    flaggerv1.CanaryDeploymentKind,
+					}),
+				},
+			},
+			Spec: appsv1.DeploymentSpec{
+				Replicas: canaryDep.Spec.Replicas,
+				Strategy: canaryDep.Spec.Strategy,
+				Selector: &metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						"app": primaryName,
+					},
+				},
+				Template: corev1.PodTemplateSpec{
+					ObjectMeta: metav1.ObjectMeta{
+						Labels:      map[string]string{"app": primaryName},
+						Annotations: canaryDep.Spec.Template.Annotations,
+					},
+					Spec: canaryDep.Spec.Template.Spec,
+				},
+			},
+		}
+
+		_, err = c.kubeClient.AppsV1().Deployments(cd.Namespace).Create(primaryDep)
+		if err != nil {
+			return err
+		}
+
+		c.recordEventInfof(cd, "Deployment %s.%s created", primaryDep.GetName(), cd.Namespace)
+	}
+
+	if cd.Status.State == "" {
+		c.scaleToZeroCanary(cd)
+	}
+
+	canaryService, err := c.kubeClient.CoreV1().Services(cd.Namespace).Get(canaryName, metav1.GetOptions{})
+	if errors.IsNotFound(err) {
+		canaryService = &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      canaryName,
+				Namespace: cd.Namespace,
+				OwnerReferences: []metav1.OwnerReference{
+					*metav1.NewControllerRef(cd, schema.GroupVersionKind{
+						Group:   flaggerv1.SchemeGroupVersion.Group,
+						Version: flaggerv1.SchemeGroupVersion.Version,
+						Kind:    flaggerv1.CanaryDeploymentKind,
+					}),
+				},
+			},
+			Spec: corev1.ServiceSpec{
+				Type:     corev1.ServiceTypeClusterIP,
+				Selector: map[string]string{"app": canaryName},
+				Ports: []corev1.ServicePort{
+					{
+						Name:     "http",
+						Protocol: corev1.ProtocolTCP,
+						Port:     cd.Spec.Service.Port,
+						TargetPort: intstr.IntOrString{
+							Type:   intstr.Int,
+							IntVal: cd.Spec.Service.Port,
+						},
+					},
+				},
+			},
+		}
+
+		_, err = c.kubeClient.CoreV1().Services(cd.Namespace).Create(canaryService)
+		if err != nil {
+			return err
+		}
+		c.recordEventInfof(cd, "Service %s.%s created", canaryService.GetName(), cd.Namespace)
+	}
+
+	canaryTestServiceName := fmt.Sprintf("%s-canary", cd.Spec.TargetRef.Name)
+	canaryTestService, err := c.kubeClient.CoreV1().Services(cd.Namespace).Get(canaryTestServiceName, metav1.GetOptions{})
+	if errors.IsNotFound(err) {
+		canaryTestService = &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      canaryTestServiceName,
+				Namespace: cd.Namespace,
+				OwnerReferences: []metav1.OwnerReference{
+					*metav1.NewControllerRef(cd, schema.GroupVersionKind{
+						Group:   flaggerv1.SchemeGroupVersion.Group,
+						Version: flaggerv1.SchemeGroupVersion.Version,
+						Kind:    flaggerv1.CanaryDeploymentKind,
+					}),
+				},
+			},
+			Spec: corev1.ServiceSpec{
+				Type:     corev1.ServiceTypeClusterIP,
+				Selector: map[string]string{"app": canaryName},
+				Ports: []corev1.ServicePort{
+					{
+						Name:     "http",
+						Protocol: corev1.ProtocolTCP,
+						Port:     cd.Spec.Service.Port,
+						TargetPort: intstr.IntOrString{
+							Type:   intstr.Int,
+							IntVal: cd.Spec.Service.Port,
+						},
+					},
+				},
+			},
+		}
+
+		_, err = c.kubeClient.CoreV1().Services(cd.Namespace).Create(canaryTestService)
+		if err != nil {
+			return err
+		}
+		c.recordEventInfof(cd, "Service %s.%s created", canaryTestService.GetName(), cd.Namespace)
+	}
+
+	primaryService, err := c.kubeClient.CoreV1().Services(cd.Namespace).Get(primaryName, metav1.GetOptions{})
+	if errors.IsNotFound(err) {
+		primaryService = &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      primaryName,
+				Namespace: cd.Namespace,
+				OwnerReferences: []metav1.OwnerReference{
+					*metav1.NewControllerRef(cd, schema.GroupVersionKind{
+						Group:   flaggerv1.SchemeGroupVersion.Group,
+						Version: flaggerv1.SchemeGroupVersion.Version,
+						Kind:    flaggerv1.CanaryDeploymentKind,
+					}),
+				},
+			},
+			Spec: corev1.ServiceSpec{
+				Type:     corev1.ServiceTypeClusterIP,
+				Selector: map[string]string{"app": primaryName},
+				Ports: []corev1.ServicePort{
+					{
+						Name:     "http",
+						Protocol: corev1.ProtocolTCP,
+						Port:     cd.Spec.Service.Port,
+						TargetPort: intstr.IntOrString{
+							Type:   intstr.Int,
+							IntVal: cd.Spec.Service.Port,
+						},
+					},
+				},
+			},
+		}
+
+		_, err = c.kubeClient.CoreV1().Services(cd.Namespace).Create(primaryService)
+		if err != nil {
+			return err
+		}
+
+		c.recordEventInfof(cd, "Service %s.%s created", primaryService.GetName(), cd.Namespace)
+	}
+
+	hosts := append(cd.Spec.Service.Hosts, canaryName)
+	gateways := append(cd.Spec.Service.Gateways, "mesh")
+	virtualService, err := c.istioClient.NetworkingV1alpha3().VirtualServices(cd.Namespace).Get(canaryName, metav1.GetOptions{})
+	if errors.IsNotFound(err) {
+		virtualService = &istiov1alpha3.VirtualService{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      canaryName,
+				Namespace: cd.Namespace,
+				OwnerReferences: []metav1.OwnerReference{
+					*metav1.NewControllerRef(cd, schema.GroupVersionKind{
+						Group:   flaggerv1.SchemeGroupVersion.Group,
+						Version: flaggerv1.SchemeGroupVersion.Version,
+						Kind:    flaggerv1.CanaryDeploymentKind,
+					}),
+				},
+			},
+			Spec: istiov1alpha3.VirtualServiceSpec{
+				Hosts:    hosts,
+				Gateways: gateways,
+				Http: []istiov1alpha3.HTTPRoute{
+					{
+						Route: []istiov1alpha3.DestinationWeight{
+							{
+								Destination: istiov1alpha3.Destination{
+									Host: primaryName,
+									Port: istiov1alpha3.PortSelector{
+										Number: uint32(cd.Spec.Service.Port),
+									},
+								},
+								Weight: 100,
+							},
+							{
+								Destination: istiov1alpha3.Destination{
+									Host: canaryName,
+									Port: istiov1alpha3.PortSelector{
+										Number: uint32(cd.Spec.Service.Port),
+									},
+								},
+								Weight: 0,
+							},
+						},
+					},
+				},
+			},
+		}
+
+		_, err = c.istioClient.NetworkingV1alpha3().VirtualServices(cd.Namespace).Create(virtualService)
+		if err != nil {
+			return err
+		}
+		c.recordEventInfof(cd, "VirtualService %s.%s created", virtualService.GetName(), cd.Namespace)
+	}
+
+	return nil
+}

pkg/operator/deployment.go

@@ -0,0 +1,422 @@
+package operator
+
+import (
+	"fmt"
+	"time"
+
+	istiov1alpha3 "github.com/knative/pkg/apis/istio/v1alpha3"
+	flaggerv1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1beta1"
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func (c *Controller) doRollouts() {
+	c.rollouts.Range(func(key interface{}, value interface{}) bool {
+		r := value.(*flaggerv1.CanaryDeployment)
+		if r.Spec.TargetRef.Kind == "Deployment" {
+			go c.advanceDeploymentRollout(r.Name, r.Namespace)
+		}
+		return true
+	})
+}
+
+func (c *Controller) advanceDeploymentRollout(name string, namespace string) {
+	// gate stage: check if the rollout exists
+	r, ok := c.getRollout(name, namespace)
+	if !ok {
+		return
+	}
+
+	err := c.bootstrapDeployment(r)
+	if err != nil {
+		c.recordEventWarningf(r, "%v", err)
+		return
+	}
+
+	// set max weight default value to 100%
+	maxWeight := 100
+	if r.Spec.CanaryAnalysis.MaxWeight > 0 {
+		maxWeight = r.Spec.CanaryAnalysis.MaxWeight
+	}
+
+	// gate stage: check if canary deployment exists and is healthy
+	canary, ok := c.getCanaryDeployment(r, r.Spec.TargetRef.Name, r.Namespace)
+	if !ok {
+		return
+	}
+
+	// gate stage: check if primary deployment exists and is healthy
+	primary, ok := c.getDeployment(r, fmt.Sprintf("%s-primary", r.Spec.TargetRef.Name), r.Namespace)
+	if !ok {
+		return
+	}
+
+	// gate stage: check if virtual service exists
+	// and if it contains weighted destination routes to the primary and canary services
+	vs, primaryRoute, canaryRoute, ok := c.getVirtualService(r)
+	if !ok {
+		return
+	}
+
+	// gate stage: check if rollout should start (canary revision has changes) or continue
+	if ok := c.checkRolloutStatus(r, canary); !ok {
+		return
+	}
+
+	// gate stage: check if the number of failed checks reached the threshold
+	if r.Status.State == "running" && r.Status.FailedChecks >= r.Spec.CanaryAnalysis.Threshold {
+		c.recordEventWarningf(r, "Rolling back %s.%s failed checks threshold reached %v",
+			r.Name, r.Namespace, r.Status.FailedChecks)
+
+		// route all traffic back to primary
+		primaryRoute.Weight = 100
+		canaryRoute.Weight = 0
+		if ok := c.updateVirtualServiceRoutes(r, vs, primaryRoute, canaryRoute); !ok {
+			return
+		}
+
+		c.recordEventWarningf(r, "Canary failed! Scaling down %s.%s",
+			canary.GetName(), canary.Namespace)
+
+		// shutdown canary
+		c.scaleToZeroCanary(r)
+
+		// mark rollout as failed
+		c.updateRolloutStatus(r, "promotion-failed")
+		return
+	}
+
+	// gate stage: check if the canary success rate is above the threshold
+	// skip check if no traffic is routed to canary
+	if canaryRoute.Weight == 0 {
+		c.recordEventInfof(r, "Starting canary deployment for %s.%s", r.Name, r.Namespace)
+	} else {
+		if ok := c.checkDeploymentMetrics(r); !ok {
+			c.updateRolloutFailedChecks(r, r.Status.FailedChecks+1)
+			return
+		}
+	}
+
+	// routing stage: increase canary traffic percentage
+	if canaryRoute.Weight < maxWeight {
+		primaryRoute.Weight -= r.Spec.CanaryAnalysis.StepWeight
+		if primaryRoute.Weight < 0 {
+			primaryRoute.Weight = 0
+		}
+		canaryRoute.Weight += r.Spec.CanaryAnalysis.StepWeight
+		if primaryRoute.Weight > 100 {
+			primaryRoute.Weight = 100
+		}
+
+		if ok := c.updateVirtualServiceRoutes(r, vs, primaryRoute, canaryRoute); !ok {
+			return
+		}
+
+		c.recordEventInfof(r, "Advance %s.%s canary weight %v", r.Name, r.Namespace, canaryRoute.Weight)
+
+		// promotion stage: override primary.template.spec with the canary spec
+		if canaryRoute.Weight == maxWeight {
+			c.recordEventInfof(r, "Copying %s.%s template spec to %s.%s",
+				canary.GetName(), canary.Namespace, primary.GetName(), primary.Namespace)
+
+			primary.Spec.Template.Spec = canary.Spec.Template.Spec
+			_, err := c.kubeClient.AppsV1().Deployments(primary.Namespace).Update(primary)
+			if err != nil {
+				c.recordEventErrorf(r, "Updating template spec %s.%s failed: %v", primary.GetName(), primary.Namespace, err)
+				return
+			}
+		}
+	} else {
+		// final stage: route all traffic back to primary
+		primaryRoute.Weight = 100
+		canaryRoute.Weight = 0
+		if ok := c.updateVirtualServiceRoutes(r, vs, primaryRoute, canaryRoute); !ok {
+			return
+		}
+
+		// final stage: mark rollout as finished and scale canary to zero replicas
+		c.recordEventInfof(r, "Scaling down %s.%s", canary.GetName(), canary.Namespace)
+		c.scaleToZeroCanary(r)
+		c.updateRolloutStatus(r, "promotion-finished")
+	}
+}
+
+func (c *Controller) getRollout(name string, namespace string) (*flaggerv1.CanaryDeployment, bool) {
+	r, err := c.rolloutClient.FlaggerV1beta1().CanaryDeployments(namespace).Get(name, v1.GetOptions{})
+	if err != nil {
+		c.logger.Errorf("CanaryDeployment %s.%s not found", name, namespace)
+		return nil, false
+	}
+
+	return r, true
+}
+
+func (c *Controller) checkRolloutStatus(r *flaggerv1.CanaryDeployment, canary *appsv1.Deployment) bool {
+	canaryRevision, err := c.getDeploymentSpecEnc(canary)
+	if err != nil {
+		c.logger.Errorf("Canary %s.%s not found: %v", r.Name, r.Namespace, err)
+		return false
+	}
+
+	if r.Status.State == "" {
+		r.Status = flaggerv1.CanaryDeploymentStatus{
+			State:          "initialized",
+			CanaryRevision: canaryRevision,
+			FailedChecks:   0,
+		}
+		r, err = c.rolloutClient.FlaggerV1beta1().CanaryDeployments(r.Namespace).Update(r)
+		if err != nil {
+			c.logger.Errorf("Canary %s.%s status update failed: %v", r.Name, r.Namespace, err)
+			return false
+		}
+
+		c.recordEventInfof(r, "Initialization done! %s.%ss", canary.GetName(), canary.Namespace)
+		return false
+	}
+
+	if r.Status.State == "running" {
+		return true
+	}
+
+	if r.Status.State == "promotion-finished" {
+		c.setCanaryRevision(r, canary, "finished")
+		c.logger.Infof("Promotion completed! %s.%s", r.Name, r.Namespace)
+		return false
+	}
+
+	if r.Status.State == "promotion-failed" {
+		c.setCanaryRevision(r, canary, "failed")
+		c.logger.Infof("Promotion failed! %s.%s", r.Name, r.Namespace)
+		return false
+	}
+
+	if diff, err := c.diffDeploymentSpec(r, canary); diff {
+		c.recordEventInfof(r, "New revision detected %s.%s",
+			canary.GetName(), canary.Namespace)
+		canary.Spec.Replicas = int32p(1)
+		canary, err = c.kubeClient.AppsV1().Deployments(canary.Namespace).Update(canary)
+		if err != nil {
+			c.recordEventErrorf(r, "Scaling up %s.%s failed: %v", canary.GetName(), canary.Namespace, err)
+			return false
+		}
+
+		r.Status = flaggerv1.CanaryDeploymentStatus{
+			FailedChecks:   0,
+		}
+		c.setCanaryRevision(r, canary, "running")
+		c.recordEventInfof(r, "Scaling up %s.%s", canary.GetName(), canary.Namespace)
+
+		return false
+	}
+
+	return false
+}
+
+func (c *Controller) updateRolloutStatus(r *flaggerv1.CanaryDeployment, status string) bool {
+	var err error
+	r.Status.State = status
+	r, err = c.rolloutClient.FlaggerV1beta1().CanaryDeployments(r.Namespace).Update(r)
+	if err != nil {
+		c.logger.Errorf("Canary %s.%s status update failed: %v", r.Name, r.Namespace, err)
+		return false
+	}
+	return true
+}
+
+func (c *Controller) updateRolloutFailedChecks(r *flaggerv1.CanaryDeployment, val int) bool {
+	var err error
+	r.Status.FailedChecks = val
+	r, err = c.rolloutClient.FlaggerV1beta1().CanaryDeployments(r.Namespace).Update(r)
+	if err != nil {
+		c.logger.Errorf("Canary %s.%s status update failed: %v", r.Name, r.Namespace, err)
+		return false
+	}
+	return true
+}
+
+func (c *Controller) getDeployment(r *flaggerv1.CanaryDeployment, name string, namespace string) (*appsv1.Deployment, bool) {
+	dep, err := c.kubeClient.AppsV1().Deployments(namespace).Get(name, v1.GetOptions{})
+	if err != nil {
+		c.recordEventErrorf(r, "Deployment %s.%s not found", name, namespace)
+		return nil, false
+	}
+
+	if msg, healthy := getDeploymentStatus(dep); !healthy {
+		c.recordEventWarningf(r, "Halt %s.%s advancement %s", dep.GetName(), dep.Namespace, msg)
+		return nil, false
+	}
+
+	if dep.Spec.Replicas == nil || *dep.Spec.Replicas == 0 {
+		return nil, false
+	}
+
+	return dep, true
+}
+
+func (c *Controller) getCanaryDeployment(r *flaggerv1.CanaryDeployment, name string, namespace string) (*appsv1.Deployment, bool) {
+	dep, err := c.kubeClient.AppsV1().Deployments(namespace).Get(name, v1.GetOptions{})
+	if err != nil {
+		c.recordEventErrorf(r, "Deployment %s.%s not found", name, namespace)
+		return nil, false
+	}
+
+	if msg, healthy := getDeploymentStatus(dep); !healthy {
+		c.recordEventWarningf(r, "Halt %s.%s advancement %s", dep.GetName(), dep.Namespace, msg)
+		return nil, false
+	}
+
+	return dep, true
+}
+
+func (c *Controller) checkDeploymentMetrics(r *flaggerv1.CanaryDeployment) bool {
+	for _, metric := range r.Spec.CanaryAnalysis.Metrics {
+		if metric.Name == "istio_requests_total" {
+			val, err := c.getDeploymentCounter(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
+			if err != nil {
+				c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.metricsServer, err)
+				return false
+			}
+			if float64(metric.Threshold) > val {
+				c.recordEventWarningf(r, "Halt %s.%s advancement success rate %.2f%% < %v%%",
+					r.Name, r.Namespace, val, metric.Threshold)
+				return false
+			}
+		}
+
+		if metric.Name == "istio_request_duration_seconds_bucket" {
+			val, err := c.GetDeploymentHistogram(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
+			if err != nil {
+				c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.metricsServer, err)
+				return false
+			}
+			t := time.Duration(metric.Threshold) * time.Millisecond
+			if val > t {
+				c.recordEventWarningf(r, "Halt %s.%s advancement request duration %v > %v",
+					r.Name, r.Namespace, val, t)
+				return false
+			}
+		}
+	}
+
+	return true
+}
+
+func (c *Controller) scaleToZeroCanary(r *flaggerv1.CanaryDeployment) {
+	canary, err := c.kubeClient.AppsV1().Deployments(r.Namespace).Get(r.Spec.TargetRef.Name, v1.GetOptions{})
+	if err != nil {
+		c.recordEventErrorf(r, "Deployment %s.%s not found", r.Spec.TargetRef.Name, r.Namespace)
+		return
+	}
+	//HPA https://github.com/kubernetes/kubernetes/pull/29212
+	canary.Spec.Replicas = int32p(0)
+	canary, err = c.kubeClient.AppsV1().Deployments(canary.Namespace).Update(canary)
+	if err != nil {
+		c.recordEventErrorf(r, "Scaling down %s.%s failed: %v", canary.GetName(), canary.Namespace, err)
+		return
+	}
+}
+
+func (c *Controller) setCanaryRevision(r *flaggerv1.CanaryDeployment, canary *appsv1.Deployment, status string) {
+	r.Status = flaggerv1.CanaryDeploymentStatus{
+		State:        status,
+		FailedChecks: r.Status.FailedChecks,
+	}
+	err := c.saveDeploymentSpec(r, canary)
+	if err != nil {
+		c.logger.Errorf("CanaryDeployment %s.%s status update failed: %v", r.Name, r.Namespace, err)
+	}
+}
+
+func (c *Controller) getVirtualService(r *flaggerv1.CanaryDeployment) (
+	vs *istiov1alpha3.VirtualService,
+	primary istiov1alpha3.DestinationWeight,
+	canary istiov1alpha3.DestinationWeight,
+	ok bool,
+) {
+	var err error
+	vs, err = c.istioClient.NetworkingV1alpha3().VirtualServices(r.Namespace).Get(r.Name, v1.GetOptions{})
+	if err != nil {
+		c.recordEventErrorf(r, "VirtualService %s.%s not found", r.Name, r.Namespace)
+		return
+	}
+
+	for _, http := range vs.Spec.Http {
+		for _, route := range http.Route {
+			if route.Destination.Host == fmt.Sprintf("%s-primary", r.Spec.TargetRef.Name) {
+				primary = route
+			}
+			if route.Destination.Host == r.Spec.TargetRef.Name {
+				canary = route
+			}
+		}
+	}
+
+	if primary.Weight == 0 && canary.Weight == 0 {
+		c.recordEventErrorf(r, "VirtualService %s.%s does not contain routes for %s and %s",
+			r.Name, r.Namespace, fmt.Sprintf("%s-primary", r.Spec.TargetRef.Name), r.Spec.TargetRef.Name)
+		return
+	}
+
+	ok = true
+	return
+}
+
+func (c *Controller) updateVirtualServiceRoutes(
+	r *flaggerv1.CanaryDeployment,
+	vs *istiov1alpha3.VirtualService,
+	primary istiov1alpha3.DestinationWeight,
+	canary istiov1alpha3.DestinationWeight,
+) bool {
+	vs.Spec.Http = []istiov1alpha3.HTTPRoute{
+		{
+			Route: []istiov1alpha3.DestinationWeight{primary, canary},
+		},
+	}
+
+	var err error
+	vs, err = c.istioClient.NetworkingV1alpha3().VirtualServices(r.Namespace).Update(vs)
+	if err != nil {
+		c.recordEventErrorf(r, "VirtualService %s.%s update failed: %v", r.Name, r.Namespace, err)
+		return false
+	}
+	return true
+}
+
+func getDeploymentStatus(deployment *appsv1.Deployment) (string, bool) {
+	if deployment.Generation <= deployment.Status.ObservedGeneration {
+		cond := getDeploymentCondition(deployment.Status, appsv1.DeploymentProgressing)
+		if cond != nil && cond.Reason == "ProgressDeadlineExceeded" {
+			return fmt.Sprintf("deployment %q exceeded its progress deadline", deployment.GetName()), false
+		} else if deployment.Spec.Replicas != nil && deployment.Status.UpdatedReplicas < *deployment.Spec.Replicas {
+			return fmt.Sprintf("waiting for rollout to finish: %d out of %d new replicas have been updated",
+				deployment.Status.UpdatedReplicas, *deployment.Spec.Replicas), false
+		} else if deployment.Status.Replicas > deployment.Status.UpdatedReplicas {
+			return fmt.Sprintf("waiting for rollout to finish: %d old replicas are pending termination",
+				deployment.Status.Replicas-deployment.Status.UpdatedReplicas), false
+		} else if deployment.Status.AvailableReplicas < deployment.Status.UpdatedReplicas {
+			return fmt.Sprintf("waiting for rollout to finish: %d of %d updated replicas are available",
+				deployment.Status.AvailableReplicas, deployment.Status.UpdatedReplicas), false
+		}
+	} else {
+		return "waiting for rollout to finish: observed deployment generation less then desired generation", false
+	}
+
+	return "ready", true
+}
+
+func getDeploymentCondition(
+	status appsv1.DeploymentStatus,
+	conditionType appsv1.DeploymentConditionType,
+) *appsv1.DeploymentCondition {
+	for i := range status.Conditions {
+		c := status.Conditions[i]
+		if c.Type == conditionType {
+			return &c
+		}
+	}
+	return nil
+}
+
+func int32p(i int32) *int32 {
+	return &i
+}

pkg/operator/observer.go

@@ -0,0 +1,172 @@
+package operator
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+)
+
+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 *Controller) 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
+}
+
+// istio_requests_total
+func (c *Controller) getDeploymentCounter(name string, namespace string, metric string, interval string) (float64, error) {
+	var rate *float64
+	querySt := url.QueryEscape(`sum(rate(` +
+		metric + `{reporter="destination",destination_workload_namespace=~"` +
+		namespace + `",destination_workload=~"` +
+		name + `",response_code!~"5.*"}[1m])) / 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
+}
+
+// istio_request_duration_seconds_bucket
+func (c *Controller) GetDeploymentHistogram(name string, namespace string, metric string, interval string) (time.Duration, error) {
+	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
+}
+
+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/operator/operator.go

@@ -0,0 +1,229 @@
+package operator
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	istioclientset "github.com/knative/pkg/client/clientset/versioned"
+	flaggerv1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1beta1"
+	clientset "github.com/stefanprodan/flagger/pkg/client/clientset/versioned"
+	flaggerscheme "github.com/stefanprodan/flagger/pkg/client/clientset/versioned/scheme"
+	flaggerinformers "github.com/stefanprodan/flagger/pkg/client/informers/externalversions/flagger/v1beta1"
+	flaggerlisters "github.com/stefanprodan/flagger/pkg/client/listers/flagger/v1beta1"
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
+)
+
+const controllerAgentName = "flagger"
+
+type Controller struct {
+	kubeClient    kubernetes.Interface
+	istioClient   istioclientset.Interface
+	rolloutClient clientset.Interface
+	rolloutLister flaggerlisters.CanaryDeploymentLister
+	rolloutSynced cache.InformerSynced
+	rolloutWindow time.Duration
+	workqueue     workqueue.RateLimitingInterface
+	recorder      record.EventRecorder
+	logger        *zap.SugaredLogger
+	metricsServer string
+	rollouts      *sync.Map
+}
+
+func NewController(
+	kubeClient kubernetes.Interface,
+	istioClient istioclientset.Interface,
+	rolloutClient clientset.Interface,
+	rolloutInformer flaggerinformers.CanaryDeploymentInformer,
+	rolloutWindow time.Duration,
+	metricServer string,
+	logger *zap.SugaredLogger,
+
+) *Controller {
+	logger.Debug("Creating event broadcaster")
+	flaggerscheme.AddToScheme(scheme.Scheme)
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(logger.Named("event-broadcaster").Debugf)
+	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{
+		Interface: kubeClient.CoreV1().Events(""),
+	})
+	recorder := eventBroadcaster.NewRecorder(
+		scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
+
+	ctrl := &Controller{
+		kubeClient:    kubeClient,
+		istioClient:   istioClient,
+		rolloutClient: rolloutClient,
+		rolloutLister: rolloutInformer.Lister(),
+		rolloutSynced: rolloutInformer.Informer().HasSynced,
+		workqueue:     workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), controllerAgentName),
+		recorder:      recorder,
+		logger:        logger,
+		rollouts:      new(sync.Map),
+		metricsServer: metricServer,
+		rolloutWindow: rolloutWindow,
+	}
+
+	rolloutInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: ctrl.enqueue,
+		UpdateFunc: func(old, new interface{}) {
+			oldRoll, ok := checkCustomResourceType(old, logger)
+			if !ok {
+				return
+			}
+			newRoll, ok := checkCustomResourceType(new, logger)
+			if !ok {
+				return
+			}
+
+			if diff := cmp.Diff(newRoll.Spec, oldRoll.Spec); diff != "" {
+				ctrl.logger.Debugf("Diff detected %s.%s %s", oldRoll.Name, oldRoll.Namespace, diff)
+				ctrl.enqueue(new)
+			}
+		},
+		DeleteFunc: func(old interface{}) {
+			r, ok := checkCustomResourceType(old, logger)
+			if ok {
+				ctrl.logger.Infof("Deleting %s.%s from cache", r.Name, r.Namespace)
+				ctrl.rollouts.Delete(fmt.Sprintf("%s.%s", r.Name, r.Namespace))
+			}
+		},
+	})
+
+	return ctrl
+}
+
+func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
+	defer utilruntime.HandleCrash()
+	defer c.workqueue.ShutDown()
+
+	c.logger.Info("Starting operator")
+
+	for i := 0; i < threadiness; i++ {
+		go wait.Until(func() {
+			for c.processNextWorkItem() {
+			}
+		}, time.Second, stopCh)
+	}
+
+	c.logger.Info("Started operator workers")
+
+	tickChan := time.NewTicker(c.rolloutWindow).C
+	for {
+		select {
+		case <-tickChan:
+			c.doRollouts()
+		case <-stopCh:
+			c.logger.Info("Shutting down operator workers")
+			return nil
+		}
+	}
+
+	return nil
+}
+
+func (c *Controller) processNextWorkItem() bool {
+	obj, shutdown := c.workqueue.Get()
+
+	if shutdown {
+		return false
+	}
+
+	err := func(obj interface{}) error {
+		defer c.workqueue.Done(obj)
+		var key string
+		var ok bool
+		if key, ok = obj.(string); !ok {
+			c.workqueue.Forget(obj)
+			utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
+			return nil
+		}
+		// Run the syncHandler, passing it the namespace/name string of the
+		// Foo resource to be synced.
+		if err := c.syncHandler(key); err != nil {
+			return fmt.Errorf("error syncing '%s': %s", key, err.Error())
+		}
+		// Finally, if no error occurs we Forget this item so it does not
+		// get queued again until another change happens.
+		c.workqueue.Forget(obj)
+		return nil
+	}(obj)
+
+	if err != nil {
+		utilruntime.HandleError(err)
+		return true
+	}
+
+	return true
+}
+
+func (c *Controller) syncHandler(key string) error {
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
+		return nil
+	}
+	cd, err := c.rolloutLister.CanaryDeployments(namespace).Get(name)
+	if errors.IsNotFound(err) {
+		utilruntime.HandleError(fmt.Errorf("'%s' in work queue no longer exists", key))
+		return nil
+	}
+
+	c.rollouts.Store(fmt.Sprintf("%s.%s", cd.Name, cd.Namespace), cd)
+
+	err = c.bootstrapDeployment(cd)
+	if err != nil {
+		c.logger.Warnf("%s.%s bootstrap error %v", cd.Name, cd.Namespace, err)
+		return err
+	}
+
+	c.logger.Infof("Synced %s", key)
+
+	return nil
+}
+
+func (c *Controller) enqueue(obj interface{}) {
+	var key string
+	var err error
+	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+	c.workqueue.AddRateLimited(key)
+}
+
+func checkCustomResourceType(obj interface{}, logger *zap.SugaredLogger) (flaggerv1.CanaryDeployment, bool) {
+	var roll *flaggerv1.CanaryDeployment
+	var ok bool
+	if roll, ok = obj.(*flaggerv1.CanaryDeployment); !ok {
+		logger.Errorf("Event Watch received an invalid object: %#v", obj)
+		return flaggerv1.CanaryDeployment{}, false
+	}
+	return *roll, true
+}
+
+func (c *Controller) recordEventInfof(r *flaggerv1.CanaryDeployment, template string, args ...interface{}) {
+	c.logger.Infof(template, args...)
+	c.recorder.Event(r, corev1.EventTypeNormal, "Synced", fmt.Sprintf(template, args...))
+}
+
+func (c *Controller) recordEventErrorf(r *flaggerv1.CanaryDeployment, template string, args ...interface{}) {
+	c.logger.Errorf(template, args...)
+	c.recorder.Event(r, corev1.EventTypeWarning, "Synced", fmt.Sprintf(template, args...))
+}
+
+func (c *Controller) recordEventWarningf(r *flaggerv1.CanaryDeployment, template string, args ...interface{}) {
+	c.logger.Infof(template, args...)
+	c.recorder.Event(r, corev1.EventTypeWarning, "Synced", fmt.Sprintf(template, args...))
+}

pkg/operator/utils.go

@@ -0,0 +1,79 @@
+package operator
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	flaggerv1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1beta1"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func (c *Controller) saveDeploymentSpec(cd *flaggerv1.CanaryDeployment, dep *appsv1.Deployment) error {
+	specJson, err := json.Marshal(dep.Spec.Template.Spec)
+	if err != nil {
+		return err
+	}
+
+	specEnc := base64.StdEncoding.EncodeToString(specJson)
+	cd.Status.CanaryRevision = specEnc
+	cd, err = c.rolloutClient.FlaggerV1beta1().CanaryDeployments(cd.Namespace).Update(cd)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *Controller) diffDeploymentSpec(cd *flaggerv1.CanaryDeployment, dep *appsv1.Deployment) (bool, error) {
+	if cd.Status.CanaryRevision == "" {
+		return true, nil
+	}
+
+	newSpec := &dep.Spec.Template.Spec
+	oldSpecJson, err := base64.StdEncoding.DecodeString(cd.Status.CanaryRevision)
+	if err != nil {
+		return false, err
+	}
+	oldSpec := &corev1.PodSpec{}
+	err = json.Unmarshal(oldSpecJson, oldSpec)
+	if err != nil {
+		return false, err
+	}
+
+	if diff := cmp.Diff(*newSpec, *oldSpec, cmpopts.IgnoreUnexported(resource.Quantity{})); diff != "" {
+		fmt.Println(diff)
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func (c *Controller) getDeploymentSpec(name string, namespace string) (string, error) {
+	dep, err := c.kubeClient.AppsV1().Deployments(namespace).Get(name, v1.GetOptions{})
+	if err != nil {
+		return "", err
+	}
+
+	specJson, err := json.Marshal(dep.Spec.Template.Spec)
+	if err != nil {
+		return "", err
+	}
+
+	specEnc := base64.StdEncoding.EncodeToString(specJson)
+	return specEnc, nil
+}
+
+func (c *Controller) getDeploymentSpecEnc(dep *appsv1.Deployment) (string, error) {
+	specJson, err := json.Marshal(dep.Spec.Template.Spec)
+	if err != nil {
+		return "", err
+	}
+
+	specEnc := base64.StdEncoding.EncodeToString(specJson)
+	return specEnc, nil
+}