Merge pull request #56 from platform9/kubewatcher

Kubewatcher: trigger functions from Kubernetes Watch callbacks

common.go

@@ -0,0 +1,30 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package fission
+
+import (
+	"fmt"
+)
+
+func UrlForFunction(m *Metadata) string {
+	prefix := "/fission-function"
+	if len(m.Uid) > 0 {
+		return fmt.Sprintf("%v/%v/%v", prefix, m.Name, m.Uid)
+	} else {
+		return fmt.Sprintf("%v/%v", prefix, m.Name)
+	}
+}

controller/api.go

@@ -33,6 +33,17 @@ type API struct {
 	FunctionStore
 	HTTPTriggerStore
 	EnvironmentStore
+	WatchStore
+}
+
+func MakeAPI(rs *ResourceStore) *API {
+	api := &API{
+		FunctionStore:    FunctionStore{ResourceStore: *rs},
+		HTTPTriggerStore: HTTPTriggerStore{ResourceStore: *rs},
+		EnvironmentStore: EnvironmentStore{ResourceStore: *rs},
+		WatchStore:       WatchStore{ResourceStore: *rs},
+	}
+	return api
 }
 
 func (api *API) respondWithSuccess(w http.ResponseWriter, resp []byte) {
@@ -76,6 +87,12 @@ func (api *API) Serve(port int) {
 	r.HandleFunc("/v1/environments/{environment}", api.EnvironmentApiUpdate).Methods("PUT")
 	r.HandleFunc("/v1/environments/{environment}", api.EnvironmentApiDelete).Methods("DELETE")
 
+	r.HandleFunc("/v1/watches", api.WatchApiList).Methods("GET")
+	r.HandleFunc("/v1/watches", api.WatchApiCreate).Methods("POST")
+	r.HandleFunc("/v1/watches/{watch}", api.WatchApiGet).Methods("GET")
+	r.HandleFunc("/v1/watches/{watch}", api.WatchApiUpdate).Methods("PUT")
+	r.HandleFunc("/v1/watches/{watch}", api.WatchApiDelete).Methods("DELETE")
+
 	address := fmt.Sprintf(":%v", port)
 
 	log.WithFields(log.Fields{"port": port}).Info("Server started")

controller/api_test.go

@@ -185,22 +185,55 @@ func TestEnvironmentApi(t *testing.T) {
 	assert(len(ts) == 2, "created two envs, but didn't find them")
 }
 
+func TestWatchApi(t *testing.T) {
+	testWatch := &fission.Watch{
+		Metadata: fission.Metadata{
+			Name: "xxx",
+			Uid:  "yyy",
+		},
+		Namespace:     "default",
+		ObjType:       "pod",
+		LabelSelector: "",
+		FieldSelector: "",
+		Function: fission.Metadata{
+			Name: "foo",
+			Uid:  "",
+		},
+		Target: "",
+	}
+	m, err := g.client.WatchCreate(testWatch)
+	panicIf(err)
+	defer g.client.WatchDelete(m)
+
+	w, err := g.client.WatchGet(m)
+	panicIf(err)
+	testWatch.Metadata.Uid = m.Uid
+	w.Target = ""
+	assert(*testWatch == *w, "watch should match after reading")
+
+	testWatch.Metadata.Name = "yyy"
+	m2, err := g.client.WatchCreate(testWatch)
+	panicIf(err)
+	defer g.client.WatchDelete(m2)
+
+	ws, err := g.client.WatchList()
+	panicIf(err)
+	assert(len(ws) == 2, "created two envs, but didn't find them")
+}
+
 func TestMain(m *testing.M) {
 	flag.Parse()
 
 	fileStore, ks, rs := getTestResourceStore()
 	defer os.RemoveAll(fileStore.root)
 
-	api := &API{
-		FunctionStore:    FunctionStore{ResourceStore: *rs},
-		HTTPTriggerStore: HTTPTriggerStore{ResourceStore: *rs},
-		EnvironmentStore: EnvironmentStore{ResourceStore: *rs},
-	}
+	api := MakeAPI(rs)
 	g.client = client.MakeClient("http://localhost:8888")
 
 	ks.Delete(context.Background(), "Function", &etcdClient.DeleteOptions{Recursive: true})
 	ks.Delete(context.Background(), "HTTPTrigger", &etcdClient.DeleteOptions{Recursive: true})
 	ks.Delete(context.Background(), "Environment", &etcdClient.DeleteOptions{Recursive: true})
+	ks.Delete(context.Background(), "Watch", &etcdClient.DeleteOptions{Recursive: true})
 
 	go api.Serve(8888)
 	time.Sleep(500 * time.Millisecond)

controller/client/client.go

@@ -459,3 +459,90 @@ func (c *Client) EnvironmentList() ([]fission.Environment, error) {
 
 	return envs, nil
 }
+
+func (c *Client) WatchCreate(w *fission.Watch) (*fission.Metadata, error) {
+	reqbody, err := json.Marshal(w)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := http.Post(c.url("watches"), "application/json", bytes.NewReader(reqbody))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var m fission.Metadata
+	err = json.Unmarshal(body, &m)
+	if err != nil {
+		return nil, err
+	}
+
+	return &m, nil
+}
+
+func (c *Client) WatchGet(m *fission.Metadata) (*fission.Watch, error) {
+	relativeUrl := fmt.Sprintf("watches/%v", m.Name)
+	if len(m.Uid) > 0 {
+		relativeUrl += fmt.Sprintf("?uid=%v", m.Uid)
+	}
+
+	resp, err := http.Get(c.url(relativeUrl))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var w fission.Watch
+	err = json.Unmarshal(body, &w)
+	if err != nil {
+		return nil, err
+	}
+
+	return &w, nil
+}
+
+func (c *Client) WatchUpdate(w *fission.Watch) (*fission.Metadata, error) {
+	return nil, fission.MakeError(fission.ErrorNotImplmented,
+		"watch update not implemented")
+}
+
+func (c *Client) WatchDelete(m *fission.Metadata) error {
+	relativeUrl := fmt.Sprintf("watches/%v", m.Name)
+	if len(m.Uid) > 0 {
+		relativeUrl += fmt.Sprintf("?uid=%v", m.Uid)
+	}
+	err := c.delete(relativeUrl)
+	return err
+
+}
+
+func (c *Client) WatchList() ([]fission.Watch, error) {
+	resp, err := http.Get(c.url("watches"))
+	if err != nil {
+		return nil, err
+	}
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	watches := make([]fission.Watch, 0)
+	err = json.Unmarshal(body, &watches)
+	if err != nil {
+		return nil, err
+	}
+
+	return watches, err
+}

controller/watchApi.go

@@ -0,0 +1,121 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package controller
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/gorilla/mux"
+
+	"github.com/platform9/fission"
+)
+
+func (api *API) WatchApiList(w http.ResponseWriter, r *http.Request) {
+	watches, err := api.WatchStore.List()
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	resp, err := json.Marshal(watches)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) WatchApiCreate(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		api.respondWithError(w, err)
+	}
+
+	var watch fission.Watch
+	err = json.Unmarshal(body, &watch)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	watch.Target = fission.UrlForFunction(&watch.Function)
+
+	uid, err := api.WatchStore.Create(&watch)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	m := &fission.Metadata{Name: watch.Metadata.Name, Uid: uid}
+	resp, err := json.Marshal(m)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) WatchApiGet(w http.ResponseWriter, r *http.Request) {
+	var m fission.Metadata
+
+	vars := mux.Vars(r)
+	m.Name = vars["watch"]
+	m.Uid = r.FormValue("uid") // empty if uid is absent
+
+	watch, err := api.WatchStore.Get(&m)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	resp, err := json.Marshal(watch)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) WatchApiUpdate(w http.ResponseWriter, r *http.Request) {
+	api.respondWithError(w, fission.MakeError(fission.ErrorNotImplmented,
+		"Not implemented"))
+}
+
+func (api *API) WatchApiDelete(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	var m fission.Metadata
+	m.Name = vars["watch"]
+
+	m.Uid = r.FormValue("uid") // empty if uid is absent
+	if len(m.Uid) == 0 {
+		log.WithFields(log.Fields{"watch": m.Name}).Info("Deleting all versions")
+	}
+
+	err := api.WatchStore.Delete(m)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	api.respondWithSuccess(w, []byte(""))
+}

controller/watchStore.go

@@ -0,0 +1,79 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package controller
+
+import (
+	"github.com/satori/go.uuid"
+
+	"github.com/platform9/fission"
+)
+
+type WatchStore struct {
+	ResourceStore
+}
+
+func (ws *WatchStore) Create(w *fission.Watch) (string, error) {
+	w.Metadata.Uid = uuid.NewV4().String()
+	return w.Metadata.Uid, ws.ResourceStore.create(w)
+}
+
+func (ws *WatchStore) Get(m *fission.Metadata) (*fission.Watch, error) {
+	var w fission.Watch
+	err := ws.ResourceStore.read(m.Name, &w)
+	if err != nil {
+		return nil, err
+	}
+	return &w, err
+}
+
+func (ws *WatchStore) Update(w *fission.Watch) (string, error) {
+	w.Metadata.Uid = uuid.NewV4().String()
+	return w.Metadata.Uid, ws.ResourceStore.update(w)
+}
+
+func (ws *WatchStore) Delete(m fission.Metadata) error {
+	typeName, err := getTypeName(fission.Watch{})
+	if err != nil {
+		return err
+	}
+	return ws.ResourceStore.delete(typeName, m.Name)
+}
+
+func (ws *WatchStore) List() ([]fission.Watch, error) {
+	typeName, err := getTypeName(fission.Watch{})
+	if err != nil {
+		return nil, err
+	}
+
+	bufs, err := ws.ResourceStore.getAll(typeName)
+	if err != nil {
+		return nil, err
+	}
+
+	watches := make([]fission.Watch, 0, len(bufs))
+	js := JsonSerializer{}
+	for _, buf := range bufs {
+		var w fission.Watch
+		err = js.deserialize([]byte(buf), &w)
+		if err != nil {
+			return nil, err
+		}
+		watches = append(watches, w)
+	}
+
+	return watches, nil
+}

environments/nodejs/server.js

@@ -82,6 +82,7 @@ app.all('/', function (req, res) {
     try {
         userFunction(context, callback);
     } catch(e) {
+        console.log(`Function error: ${e}`);
         callback(500, "Internal server error")
     }
 });

fission-bundle/main.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/docopt/docopt-go"
 	"github.com/platform9/fission/controller"
+	"github.com/platform9/fission/kubewatcher"
 	"github.com/platform9/fission/poolmgr"
 	"github.com/platform9/fission/router"
 )
@@ -19,11 +20,7 @@ func runController(port int, etcdUrl string, filepath string) {
 		log.Fatalf("Error: %v", err)
 	}
 
-	api := &controller.API{
-		FunctionStore:    controller.FunctionStore{ResourceStore: *rs},
-		HTTPTriggerStore: controller.HTTPTriggerStore{ResourceStore: *rs},
-		EnvironmentStore: controller.EnvironmentStore{ResourceStore: *rs},
-	}
+	api := controller.MakeAPI(rs)
 	api.Serve(port)
 	log.Fatalf("Error: Controller exited.")
 }
@@ -40,6 +37,13 @@ func runPoolmgr(port int, controllerUrl string, namespace string) {
 	}
 }
 
+func runKubeWatcher(controllerUrl, routerUrl string) {
+	err := kubewatcher.Start(controllerUrl, routerUrl)
+	if err != nil {
+		log.Fatalf("Error starting kubewatcher: %v", err)
+	}
+}
+
 func getPort(portArg interface{}) int {
 	portArgStr := portArg.(string)
 	port, err := strconv.Atoi(portArgStr)
@@ -72,15 +76,18 @@ Usage:
   fission-bundle --controllerPort=<port> [--etcdUrl=<etcdUrl>] --filepath=<filepath>
   fission-bundle --routerPort=<port> [--controllerUrl=<url> --poolmgrUrl=<url>]
   fission-bundle --poolmgrPort=<port> [--controllerUrl=<url>]
+  fission-bundle --kubewatcher [--controllerUrl=<url> --routerUrl=<url>]
 Options:
   --controllerPort=<port>  Port that the controller should listen on.
   --routerPort=<port>      Port that the router should listen on.
   --poolmgrPort=<port>     Port that the poolmgr should listen on.
   --controllerUrl=<url>    Controller URL. Not required if --controllerPort is specified.
-  --poolmgrUrl=<url>       Controller URL. Not required if --poolmgrPort is specified.
+  --poolmgrUrl=<url>       Poolmgr URL. Not required if --poolmgrPort is specified.
+  --routerUrl=<url>        Router URL.
   --etcdUrl=<etcdUrl>      Etcd URL.
   --filepath=<filepath>    Directory to store functions in.
   --namespace=<namespace>  Kubernetes namespace in which to run function containers. Defaults to 'fission-function'.
+  --kubewatcher            Start Kubernetes events watcher.
 `
 	arguments, err := docopt.Parse(usage, nil, true, "fission-bundle", false)
 	if err != nil {
@@ -92,6 +99,7 @@ Options:
 	controllerUrl := getStringArgWithDefault(arguments["--controllerUrl"], "http://controller.fission")
 	etcdUrl := getStringArgWithDefault(arguments["--etcdUrl"], "http://etcd:2379")
 	poolmgrUrl := getStringArgWithDefault(arguments["--poolmgrUrl"], "http://poolmgr.fission")
+	routerUrl := getStringArgWithDefault(arguments["--routerUrl"], "http://router.fission")
 
 	if arguments["--controllerPort"] != nil {
 		port := getPort(arguments["--controllerPort"])
@@ -108,5 +116,9 @@ Options:
 		runPoolmgr(port, controllerUrl, namespace)
 	}
 
+	if arguments["--kubewatcher"] == true {
+		runKubeWatcher(controllerUrl, routerUrl)
+	}
+
 	select {}
 }

fission.yaml

@@ -22,13 +22,13 @@ metadata:
 spec:
   replicas: 1
   template:
-    metadata: 
+    metadata:
       labels:
         svc: controller
     spec:
       containers:
       - name: controller
-        image: fission/fission-bundle:alpha
+        image: fission/fission-bundle:alpha10
         command: ["/fission-bundle"]
         args: ["--controllerPort", "8888", "--filepath", "/filestore"]
 
@@ -41,13 +41,13 @@ metadata:
 spec:
   replicas: 1
   template:
-    metadata: 
+    metadata:
       labels:
         svc: router
     spec:
       containers:
       - name: router
-        image: fission/fission-bundle:alpha
+        image: fission/fission-bundle:alpha10
         command: ["/fission-bundle"]
         args: ["--routerPort", "8888"]
 
@@ -75,16 +75,35 @@ metadata:
 spec:
   replicas: 1
   template:
-    metadata: 
+    metadata:
       labels:
         svc: poolmgr
     spec:
       containers:
       - name: poolmgr
-        image: fission/fission-bundle:alpha
+        image: fission/fission-bundle:alpha10
         command: ["/fission-bundle"]
         args: ["--poolmgrPort", "8888"]
 
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: kubewatcher
+  namespace: fission
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        svc: kubewatcher
+    spec:
+      containers:
+      - name: kubewatcher
+        image: fission/fission-bundle:alpha10
+        command: ["/fission-bundle"]
+        args: ["--kubewatcher"]
+
 ---
 apiVersion: v1
 kind: Service

fission/common.go

@@ -19,6 +19,7 @@ package main
 import (
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/platform9/fission/controller/client"
 )
@@ -34,6 +35,8 @@ func getClient(serverUrl string) *client.Client {
 		fatal("Need --server or FISSION_URL set to your fission server.")
 	}
 
+	serverUrl = "http://" + strings.TrimPrefix(serverUrl, "http://")
+
 	return client.MakeClient(serverUrl)
 }
 

fission/main.go

@@ -49,7 +49,7 @@ func main() {
 	// httptriggers
 	htNameFlag := cli.StringFlag{Name: "name", Usage: "HTTP Trigger name"}
 	htMethodFlag := cli.StringFlag{Name: "method", Usage: "HTTP Method: GET|POST|PUT|DELETE|HEAD; defaults to GET"}
-	htUrlFlag := cli.StringFlag{Name: "url", Usage: "URL pattern (see TODO for supported patterns)"}
+	htUrlFlag := cli.StringFlag{Name: "url", Usage: "URL pattern (See gorilla/mux supported patterns)"}
 	htFnNameFlag := cli.StringFlag{Name: "function", Usage: "Function name"}
 	htFnUidFlag := cli.StringFlag{Name: "uid", Usage: "Function UID (optional; uses latest if unspecified)"}
 	htSubcommands := []cli.Command{
@@ -71,10 +71,26 @@ func main() {
 		{Name: "list", Usage: "List all environments", Flags: []cli.Flag{}, Action: envList},
 	}
 
+	// watches
+	wNameFlag := cli.StringFlag{Name: "name", Usage: "Watch name"}
+	wFnNameFlag := cli.StringFlag{Name: "function", Usage: "Function name"}
+	wFnUidFlag := cli.StringFlag{Name: "uid", Usage: "Function UID (optional; uses latest if unspecified)"}
+	wNamespaceFlag := cli.StringFlag{Name: "ns", Usage: "Namespace of resource to watch"}
+	wObjTypeFlag := cli.StringFlag{Name: "type", Usage: "Type of resource to watch (Pod, Service, etc.)"}
+	wLabelsFlag := cli.StringFlag{Name: "labels", Usage: "Label selector of the form a=b,c=d"}
+	wSubCommands := []cli.Command{
+		{Name: "create", Aliases: []string{"add"}, Usage: "Create a watch", Flags: []cli.Flag{wFnNameFlag, wFnUidFlag, wNamespaceFlag, wObjTypeFlag, wLabelsFlag}, Action: wCreate},
+		{Name: "get", Usage: "Get details about a watch", Flags: []cli.Flag{wNameFlag}, Action: wGet},
+		// TODO add update flag when supported
+		{Name: "delete", Usage: "Delete watch", Flags: []cli.Flag{wNameFlag}, Action: wDelete},
+		{Name: "list", Usage: "List all watches", Flags: []cli.Flag{}, Action: wList},
+	}
+
 	app.Commands = []cli.Command{
 		{Name: "function", Aliases: []string{"fn"}, Usage: "Create, update and manage functions", Subcommands: fnSubcommands},
 		{Name: "httptrigger", Aliases: []string{"ht", "route"}, Usage: "Manage HTTP triggers (routes) for functions", Subcommands: htSubcommands},
 		{Name: "environment", Aliases: []string{"env"}, Usage: "Manage environments", Subcommands: envSubcommands},
+		{Name: "watch", Aliases: []string{"w"}, Usage: "Manage watches", Subcommands: wSubCommands},
 
 		// Misc commands
 		{

fission/watch.go

@@ -0,0 +1,120 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package main
+
+import (
+	"fmt"
+	"os"
+	"text/tabwriter"
+
+	"github.com/satori/go.uuid"
+	"github.com/urfave/cli"
+
+	"github.com/platform9/fission"
+)
+
+func wCreate(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	fnName := c.String("function")
+	if len(fnName) == 0 {
+		fatal("Need a function name to create a watch, use --function")
+	}
+	fnUid := c.String("uid")
+
+	namespace := c.String("ns")
+	if len(namespace) == 0 {
+		fmt.Println("Watch 'default' namespace. Use --ns <namespace> to override.")
+		namespace = "default"
+	}
+
+	objType := c.String("type")
+	if len(objType) == 0 {
+		fmt.Println("Object type unspecified, will watch pods.  Use --type <type> to override.")
+		objType = "pod"
+	}
+
+	labels := c.String("labels")
+	// empty 'labels' selects everything
+	if len(labels) == 0 {
+		fmt.Printf("Watching all objects of type '%v', use --labels to refine selection.\n", objType)
+	}
+
+	// automatically name watches
+	watchName := uuid.NewV4().String()
+
+	w := &fission.Watch{
+		Metadata: fission.Metadata{
+			Name: watchName,
+		},
+		Function: fission.Metadata{
+			Name: fnName,
+			Uid:  fnUid,
+		},
+		Namespace:     namespace,
+		ObjType:       objType,
+		LabelSelector: labels,
+		FieldSelector: "", // TODO
+	}
+
+	_, err := client.WatchCreate(w)
+	checkErr(err, "create watch")
+
+	fmt.Printf("watch '%v' created\n", w.Metadata.Name)
+	return err
+}
+
+func wGet(c *cli.Context) error {
+	return nil
+}
+
+func wUpdate(c *cli.Context) error {
+	return nil
+}
+
+func wDelete(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	wName := c.String("name")
+	if len(wName) == 0 {
+		fatal("Need name of watch to delete, use --name")
+	}
+
+	err := client.WatchDelete(&fission.Metadata{Name: wName})
+	checkErr(err, "delete watch")
+
+	fmt.Printf("watch '%v' deleted\n", wName)
+	return nil
+}
+
+func wList(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	ws, err := client.WatchList()
+	checkErr(err, "list watches")
+
+	w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
+
+	fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n", "NAME", "NAMESPACE", "OBJTYPE", "LABELS", "FUNCTION_NAME", "FUNCTION_UID")
+	for _, wa := range ws {
+		fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n",
+			wa.Metadata.Name, wa.Namespace, wa.ObjType, wa.LabelSelector, wa.Function.Name, wa.Function.Uid)
+	}
+	w.Flush()
+
+	return nil
+}

kubewatcher/kubewatcher.go

@@ -0,0 +1,203 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package kubewatcher
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"strings"
+	"sync/atomic"
+
+	"k8s.io/client-go/1.5/kubernetes"
+	"k8s.io/client-go/1.5/pkg/api"
+	"k8s.io/client-go/1.5/pkg/runtime"
+	"k8s.io/client-go/1.5/pkg/watch"
+
+	"github.com/platform9/fission"
+)
+
+type requestType int
+
+const (
+	SYNC requestType = iota
+)
+
+type (
+	KubeWatcher struct {
+		watches          map[string]watchSubscription
+		kubernetesClient *kubernetes.Clientset
+		requestChannel   chan *kubeWatcherRequest
+		publisher        Publisher
+	}
+
+	watchSubscription struct {
+		fission.Watch
+		kubeWatch watch.Interface
+		stopped   *int32
+	}
+
+	kubeWatcherRequest struct {
+		requestType
+		watches         []fission.Watch
+		responseChannel chan *kubeWatcherResponse
+	}
+	kubeWatcherResponse struct {
+		error
+	}
+)
+
+func MakeKubeWatcher(kubernetesClient *kubernetes.Clientset, publisher Publisher) *KubeWatcher {
+	kw := &KubeWatcher{
+		watches:          make(map[string]watchSubscription),
+		kubernetesClient: kubernetesClient,
+		publisher:        publisher,
+		requestChannel:   make(chan *kubeWatcherRequest),
+	}
+	go kw.svc()
+	return kw
+}
+
+func (kw *KubeWatcher) Sync(watches []fission.Watch) error {
+	req := &kubeWatcherRequest{
+		requestType:     SYNC,
+		watches:         watches,
+		responseChannel: make(chan *kubeWatcherResponse),
+	}
+	kw.requestChannel <- req
+	resp := <-req.responseChannel
+	return resp.error
+}
+
+func (kw *KubeWatcher) svc() {
+	for {
+		req := <-kw.requestChannel
+		switch req.requestType {
+		case SYNC:
+			newWatchUids := make(map[string]bool)
+			for _, w := range req.watches {
+				newWatchUids[w.Metadata.Uid] = true
+			}
+			// Remove old watches
+			for uid, ws := range kw.watches {
+				if _, ok := newWatchUids[uid]; !ok {
+					kw.removeWatch(&ws.Watch)
+				}
+			}
+			// Add new watches
+			for _, w := range req.watches {
+				if _, ok := kw.watches[w.Metadata.Uid]; !ok {
+					kw.addWatch(&w)
+				}
+			}
+			req.responseChannel <- &kubeWatcherResponse{error: nil}
+		}
+	}
+}
+
+// TODO lifted from kubernetes/pkg/kubectl/resource_printer.go.
+func printKubernetesObject(obj runtime.Object, w io.Writer) error {
+	switch obj := obj.(type) {
+	case *runtime.Unknown:
+		var buf bytes.Buffer
+		err := json.Indent(&buf, obj.Raw, "", "    ")
+		if err != nil {
+			return err
+		}
+		buf.WriteRune('\n')
+		_, err = buf.WriteTo(w)
+		return err
+	}
+
+	data, err := json.MarshalIndent(obj, "", "    ")
+	if err != nil {
+		return err
+	}
+	data = append(data, '\n')
+	_, err = w.Write(data)
+	return err
+}
+
+func (kw *KubeWatcher) createKubernetesWatch(w *fission.Watch) (watch.Interface, error) {
+	var wi watch.Interface
+	var err error
+
+	listOptions := api.ListOptions{} // TODO populate labelselector and fieldselector
+
+	// TODO handle the full list of types
+	switch strings.ToUpper(w.ObjType) {
+	case "POD":
+		wi, err = kw.kubernetesClient.Core().Pods(w.Namespace).Watch(listOptions)
+	case "SERVICE":
+		wi, err = kw.kubernetesClient.Core().Services(w.Namespace).Watch(listOptions)
+	default:
+		msg := fmt.Sprintf("Error: unknown obj type '%v'", w.ObjType)
+		log.Println(msg)
+		err = errors.New(msg)
+	}
+	return wi, err
+}
+
+func (kw *KubeWatcher) addWatch(w *fission.Watch) error {
+	log.Printf("Adding watch %v: %v", w.Metadata.Name, w.Function.Name)
+	wi, err := kw.createKubernetesWatch(w)
+	if err != nil {
+		return err
+	}
+	var stopped int32 = 0
+	ws := &watchSubscription{
+		Watch:     *w,
+		kubeWatch: wi,
+		stopped:   &stopped,
+	}
+	kw.watches[w.Metadata.Uid] = *ws
+	go ws.eventDispatchLoop(kw.publisher)
+	return nil
+}
+
+func (kw *KubeWatcher) removeWatch(w *fission.Watch) error {
+	log.Printf("Removing watch %v: %v", w.Metadata.Name, w.Function.Name)
+	ws, ok := kw.watches[w.Metadata.Uid]
+	if !ok {
+		return fission.MakeError(fission.ErrorNotFound,
+			fmt.Sprintf("watch doesn't exist: %v", w.Metadata))
+	}
+	delete(kw.watches, w.Metadata.Uid)
+	atomic.StoreInt32(ws.stopped, 1)
+	ws.kubeWatch.Stop()
+	return nil
+}
+
+func (ws *watchSubscription) eventDispatchLoop(publisher Publisher) {
+	log.Println("Listening to watch ", ws.Watch.Metadata.Name)
+	for {
+		ev, more := <-ws.kubeWatch.ResultChan()
+		if !more {
+			log.Println("Watch stopped", ws.Watch.Metadata.Name)
+			break
+		}
+		publisher.Publish(ev, ws.Watch.Target)
+	}
+	if atomic.LoadInt32(ws.stopped) != 0 {
+		// TODO can this happen?  How do we start the watch again from the right
+		// point?
+		log.Panicf("Watch channel closed unexpectedly")
+	}
+}

kubewatcher/main.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package kubewatcher
+
+import (
+	"log"
+
+	"k8s.io/client-go/1.5/kubernetes"
+	"k8s.io/client-go/1.5/rest"
+
+	"github.com/platform9/fission/controller/client"
+)
+
+// Get a kubernetes client using the pod's service account.
+func getKubernetesClient() (*kubernetes.Clientset, error) {
+	// creates the in-cluster config
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		log.Printf("Error getting kubernetes client config: %v", err)
+		return nil, err
+	}
+
+	// creates the clientset
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		log.Printf("Error getting kubernetes client: %v", err)
+		return nil, err
+	}
+
+	return clientset, nil
+}
+
+func Start(controllerUrl string, routerUrl string) error {
+	kubeClient, err := getKubernetesClient()
+	if err != nil {
+		return err
+	}
+	poster := MakeWebhookPublisher(routerUrl)
+	kubeWatch := MakeKubeWatcher(kubeClient, poster)
+
+	client := client.MakeClient(controllerUrl)
+	MakeWatchSync(client, kubeWatch)
+
+	return nil
+}

kubewatcher/publisher.go

@@ -0,0 +1,30 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package kubewatcher
+
+import (
+	"k8s.io/client-go/1.5/pkg/watch"
+)
+
+type (
+	Publisher interface {
+		// Publish an event to a "target".  Target's meaning depends on the
+		// publisher: it's a URL in the case of a webhook publisher, or a queue
+		// name in a queue-based publisher such as NATS.
+		Publish(event watch.Event, target string)
+	}
+)

kubewatcher/watchSync.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package kubewatcher
+
+import (
+	"log"
+	"time"
+
+	"github.com/platform9/fission/controller/client"
+)
+
+type (
+	WatchSync struct {
+		client      *client.Client
+		kubeWatcher *KubeWatcher
+	}
+)
+
+func MakeWatchSync(client *client.Client, kubeWatcher *KubeWatcher) *WatchSync {
+	ws := &WatchSync{
+		client:      client,
+		kubeWatcher: kubeWatcher,
+	}
+	go ws.syncSvc()
+	return ws
+}
+
+func (ws *WatchSync) syncSvc() {
+	failureCount := 0
+	maxFailures := 6
+	for {
+		watches, err := ws.client.WatchList()
+		if err != nil {
+			failureCount++
+			if failureCount > maxFailures {
+				log.Fatalf("Failed to connect to controller: %v", err)
+			}
+			time.Sleep(10 * time.Second)
+			continue
+		}
+
+		ws.kubeWatcher.Sync(watches)
+		time.Sleep(3 * time.Second)
+	}
+}

kubewatcher/webhookPublisher.go

@@ -0,0 +1,135 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
+package kubewatcher
+
+import (
+	"bytes"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"reflect"
+	"strings"
+	"time"
+
+	"k8s.io/client-go/1.5/pkg/watch"
+)
+
+type (
+	// A webhook publisher for a single URL. Satisifies the Publisher interface.
+	WebhookPublisher struct {
+		requestChannel chan *publishRequest
+
+		maxRetries int
+		retryDelay time.Duration
+
+		baseUrl string
+	}
+	publishRequest struct {
+		url        string
+		watchEvent watch.Event
+		retries    int
+		retryDelay time.Duration
+	}
+)
+
+func MakeWebhookPublisher(baseUrl string) *WebhookPublisher {
+	p := &WebhookPublisher{
+		baseUrl:        baseUrl,
+		requestChannel: make(chan *publishRequest, 32), // buffered channel
+		// TODO make this configurable
+		maxRetries: 10,
+		retryDelay: 500 * time.Millisecond,
+	}
+	go p.svc()
+	return p
+}
+
+func (p *WebhookPublisher) Publish(watchEvent watch.Event, url string) {
+	p.requestChannel <- &publishRequest{
+		watchEvent: watchEvent,
+		url:        url,
+		retries:    p.maxRetries,
+		retryDelay: p.retryDelay,
+	}
+}
+
+func (p *WebhookPublisher) svc() {
+	for {
+		r := <-p.requestChannel
+		p.makeHttpRequest(r)
+	}
+}
+
+func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) {
+
+	url := p.baseUrl + "/" + strings.TrimPrefix(r.url, "/")
+	log.Printf("Making HTTP request to %v", url)
+
+	// Serialize the object
+	var buf bytes.Buffer
+	err := printKubernetesObject(r.watchEvent.Object, &buf)
+	if err != nil {
+		log.Println("Failed to serialize object: %v", err)
+		// TODO send a POST request indicating error
+	}
+
+	// Create request
+	req, err := http.NewRequest("POST", url, &buf)
+	if err != nil {
+		log.Printf("Failed to create request to %v", url)
+		// can't do anything more, drop the event.
+		return
+	}
+
+	// Event and object type aren't in the serialized object
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("X-Kubernetes-Event-Type", string(r.watchEvent.Type))
+	req.Header.Add("X-Kubernetes-Object-Type", reflect.TypeOf(r.watchEvent.Object).Elem().Name())
+
+	// Make the request
+	resp, err := http.DefaultClient.Do(req)
+
+	// All done if the request succeeded with 200 OK.
+	if err == nil && resp.StatusCode == 200 {
+		resp.Body.Close()
+		return
+	}
+
+	// Log errors
+	if err != nil {
+		log.Printf("Request failed: %v", r)
+	} else if resp.StatusCode != 200 {
+		log.Printf("Request returned failure: %v", resp.StatusCode)
+		body, err := ioutil.ReadAll(resp.Body)
+		resp.Body.Close()
+		if err == nil {
+			log.Printf("request error: %v", string(body))
+		}
+	}
+
+	// Schedule a retry, or give up if out of retries
+	r.retries--
+	if r.retries > 0 {
+		r.retryDelay *= time.Duration(2)
+		time.AfterFunc(r.retryDelay, func() {
+			p.requestChannel <- r
+		})
+	} else {
+		log.Printf("Final retry failed, giving up on %v", url)
+		// Event dropped
+	}
+}

poolmgr/gp.go

@@ -443,7 +443,9 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) {
 		// our own.  TODO: this is grossly inefficient, improve it with some sort of state
 		// machine
 		log.Printf("func svc already exists: %v", existingFsvc.podName)
-		go gp.CleanupFunctionService(fsvc.podName)
+		go func() {
+			gp.kubernetesClient.Core().Pods(gp.namespace).Delete(fsvc.podName, nil)
+		}()
 		return existingFsvc, nil
 	}
 	return fsvc, nil

resource.go

@@ -1,3 +1,19 @@
+/*
+Copyright 2016 The Fission 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.
+*/
+
 package fission
 
 func (f Function) Key() string {
@@ -11,3 +27,7 @@ func (e Environment) Key() string {
 func (ht HTTPTrigger) Key() string {
 	return ht.Metadata.Name
 }
+
+func (w Watch) Key() string {
+	return w.Metadata.Name
+}

router/httpTriggers.go

@@ -35,6 +35,7 @@ type HTTPTriggerSet struct {
 	controller *controllerClient.Client
 	poolmgr    *poolmgrClient.Client
 	triggers   []fission.HTTPTrigger
+	functions  []fission.Function
 }
 
 func makeHTTPTriggerSet(fmap *functionServiceMap, controller *controllerClient.Client, poolmgr *poolmgrClient.Client) *HTTPTriggerSet {
@@ -49,12 +50,14 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, controller *controllerClient.C
 
 func (ts *HTTPTriggerSet) subscribeRouter(mr *mutableRouter) {
 	ts.mutableRouter = mr
-	mr.updateRouter(ts.getRouterFromTriggers())
+	mr.updateRouter(ts.getRouter())
 	go ts.watchTriggers()
 }
 
-func (ts *HTTPTriggerSet) getRouterFromTriggers() *mux.Router {
+func (ts *HTTPTriggerSet) getRouter() *mux.Router {
 	muxRouter := mux.NewRouter()
+
+	// HTTP triggers setup by the user
 	for _, trigger := range ts.triggers {
 		fh := &functionHandler{
 			fmap:     ts.functionServiceMap,
@@ -63,6 +66,18 @@ func (ts *HTTPTriggerSet) getRouterFromTriggers() *mux.Router {
 		}
 		muxRouter.HandleFunc(trigger.UrlPattern, fh.handler)
 	}
+
+	// Internal triggers for (the latest version of) each function
+	for _, function := range ts.functions {
+		m := fission.Metadata{Name: function.Metadata.Name}
+		fh := &functionHandler{
+			fmap:     ts.functionServiceMap,
+			Function: m,
+			poolmgr:  ts.poolmgr,
+		}
+		muxRouter.HandleFunc(fission.UrlForFunction(&m), fh.handler)
+	}
+
 	return muxRouter
 }
 
@@ -102,9 +117,18 @@ func (ts *HTTPTriggerSet) watchTriggers() {
 			if failureCount >= maxFailures {
 				log.Fatalf("Failed to connect to controller after %v retries: %v", failureCount, err)
 			}
+			time.Sleep(time.Duration(pollSleepSec) * time.Second)
+			continue
 		}
 		ts.triggers = triggers
-		ts.mutableRouter.updateRouter(ts.getRouterFromTriggers())
+
+		functions, err := ts.controller.FunctionList()
+		if err != nil {
+			log.Fatalf("Failed to get function list")
+		}
+		ts.functions = functions
+
+		ts.mutableRouter.updateRouter(ts.getRouter())
 		time.Sleep(time.Duration(pollSleepSec) * time.Second)
 	}
 }

router/router.go

@@ -43,8 +43,10 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"os"
 	"time"
 
+	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 
 	controllerClient "github.com/platform9/fission/controller/client"
@@ -65,7 +67,7 @@ func router(httpTriggerSet *HTTPTriggerSet) *mutableRouter {
 func serve(port int, httpTriggerSet *HTTPTriggerSet) {
 	mr := router(httpTriggerSet)
 	url := fmt.Sprintf(":%v", port)
-	http.ListenAndServe(url, mr)
+	http.ListenAndServe(url, handlers.LoggingHandler(os.Stdout, mr))
 }
 
 func Start(port int, controllerUrl string, poolmgrUrl string) {

types.go

@@ -53,6 +53,20 @@ type (
 		Function   Metadata `json:"function"`
 	}
 
+	// Watch is a specification of Kubernetes watch along with a URL to post events to.
+	Watch struct {
+		Metadata `json:"metadata"`
+
+		Namespace     string `json:"namespace"`
+		ObjType       string `json:"objtype"`
+		LabelSelector string `json:"labelselector"`
+		FieldSelector string `json:"fieldselector"`
+
+		Function Metadata `json:"function"`
+
+		Target string `json:"target"` // Watch publish target (URL, NATS stream, etc)
+	}
+
 	// Errors returned by the Fission API.
 	Error struct {
 		Code    errorCode `json:"code"`
@@ -69,4 +83,5 @@ const (
 	ErrorNameExists
 	ErrorInvalidArgument
 	ErrorNoSpace
+	ErrorNotImplmented
 )