Implement application watch API (#12)

pkg/apis/application/v1alpha1/types.go

@@ -2,6 +2,7 @@ package v1alpha1
 
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/rest"
 )
 
@@ -16,6 +17,18 @@ type Application struct {
 	Status            ApplicationStatus `json:"status" protobuf:"bytes,3,opt,name=status"`
 }
 
+// ApplicationWatchEvent contains information about application change.
+type ApplicationWatchEvent struct {
+	Type watch.EventType `protobuf:"bytes,1,opt,name=type,casttype=k8s.io/apimachinery/pkg/watch.EventType"`
+
+	// Application is:
+	//  * If Type is Added or Modified: the new state of the object.
+	//  * If Type is Deleted: the state of the object immediately before deletion.
+	//  * If Type is Error: *api.Status is recommended; other types may make sense
+	//    depending on context.
+	Application Application `protobuf:"bytes,2,opt,name=application"`
+}
+
 // ApplicationList is list of Application resources
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 type ApplicationList struct {

pkg/apis/application/v1alpha1/zz_generated.deepcopy.go

@@ -119,6 +119,23 @@ func (in *ApplicationStatus) DeepCopy() *ApplicationStatus {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ApplicationWatchEvent) DeepCopyInto(out *ApplicationWatchEvent) {
+	*out = *in
+	in.Application.DeepCopyInto(&out.Application)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ApplicationWatchEvent.
+func (in *ApplicationWatchEvent) DeepCopy() *ApplicationWatchEvent {
+	if in == nil {
+		return nil
+	}
+	out := new(ApplicationWatchEvent)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Cluster) DeepCopyInto(out *Cluster) {
 	*out = *in

server/application/application.go

@@ -3,6 +3,7 @@ package application
 import (
 	appv1 "github.com/argoproj/argo-cd/pkg/apis/application/v1alpha1"
 	appclientset "github.com/argoproj/argo-cd/pkg/client/clientset/versioned"
+	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -56,3 +57,29 @@ func (s *Server) ListPods(ctx context.Context, q *ApplicationQuery) (*apiv1.PodL
 	// TODO: filter by the app label
 	return s.kubeclientset.CoreV1().Pods(s.ns).List(metav1.ListOptions{})
 }
+
+func (s *Server) Watch(q *ApplicationQuery, ws ApplicationService_WatchServer) error {
+	w, err := s.appclientset.ArgoprojV1alpha1().Applications(s.ns).Watch(metav1.ListOptions{})
+	if err != nil {
+		return err
+	}
+	go func() {
+		for next := range w.ResultChan() {
+			app := *next.Object.(*appv1.Application)
+			if q.Name == "" || q.Name == app.Name {
+				err = ws.Send(&appv1.ApplicationWatchEvent{
+					Type:        next.Type,
+					Application: app,
+				})
+				if err != nil {
+					log.Warnf("Unable to send stream message: %v", err)
+				}
+			}
+		}
+	}()
+	select {
+	case <-ws.Context().Done():
+		w.Stop()
+	}
+	return nil
+}

server/application/application.proto

@@ -27,6 +27,12 @@ service ApplicationService {
 		option (google.api.http).get = "/api/v1/applications";
 	}
 
+	// Watch returns stream of application change events.
+    rpc Watch(ApplicationQuery) returns (stream github.com.argoproj.argo_cd.pkg.apis.application.v1alpha1.ApplicationWatchEvent) {
+		option (google.api.http).get = "/api/v1/stream/applications";
+	}
+
+
 	// Create creates a application
 	rpc Create(github.com.argoproj.argo_cd.pkg.apis.application.v1alpha1.Application) returns (github.com.argoproj.argo_cd.pkg.apis.application.v1alpha1.Application) {
 		option (google.api.http) = {

server/application/forwarder_overwrite.go

@@ -0,0 +1,21 @@
+package application
+
+import (
+	"net/http"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/grpc-ecosystem/grpc-gateway/runtime"
+	"golang.org/x/net/context"
+)
+
+func init() {
+	forward_ApplicationService_Watch_0 = func(ctx context.Context, mux *runtime.ServeMux, marshaler runtime.Marshaler, w http.ResponseWriter, req *http.Request, recv func() (proto.Message, error), opts ...func(context.Context, http.ResponseWriter, proto.Message) error) {
+		opts = append(opts, func(i context.Context, writer http.ResponseWriter, message proto.Message) error {
+			w.Header().Set("Content-Type", "text/event-stream")
+			w.Header().Set("Transfer-Encoding", "chunked")
+			w.Header().Set("X-Content-Type-Options", "nosniff")
+			return nil
+		})
+		runtime.ForwardResponseStream(ctx, mux, marshaler, w, req, recv, opts...)
+	}
+}