diff --git a/common.go b/common.go new file mode 100644 index 00000000..5f9a2c56 --- /dev/null +++ b/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) + } +} diff --git a/controller/api.go b/controller/api.go index f26abc29..132fd85f 100644 --- a/controller/api.go +++ b/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") diff --git a/controller/api_test.go b/controller/api_test.go index 8b218c26..f241866d 100644 --- a/controller/api_test.go +++ b/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) diff --git a/controller/client/client.go b/controller/client/client.go index 82bd7906..e73eb95a 100644 --- a/controller/client/client.go +++ b/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 +} diff --git a/controller/watchApi.go b/controller/watchApi.go new file mode 100644 index 00000000..cbec5e6b --- /dev/null +++ b/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("")) +} diff --git a/controller/watchStore.go b/controller/watchStore.go new file mode 100644 index 00000000..53f21a6e --- /dev/null +++ b/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 +} diff --git a/environments/nodejs/server.js b/environments/nodejs/server.js index e8141f9e..d40fb4e5 100644 --- a/environments/nodejs/server.js +++ b/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") } }); diff --git a/fission-bundle/main.go b/fission-bundle/main.go index d8c69fd9..26e06d97 100644 --- a/fission-bundle/main.go +++ b/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= [--etcdUrl=] --filepath= fission-bundle --routerPort= [--controllerUrl= --poolmgrUrl=] fission-bundle --poolmgrPort= [--controllerUrl=] + fission-bundle --kubewatcher [--controllerUrl= --routerUrl=] Options: --controllerPort= Port that the controller should listen on. --routerPort= Port that the router should listen on. --poolmgrPort= Port that the poolmgr should listen on. --controllerUrl= Controller URL. Not required if --controllerPort is specified. - --poolmgrUrl= Controller URL. Not required if --poolmgrPort is specified. + --poolmgrUrl= Poolmgr URL. Not required if --poolmgrPort is specified. + --routerUrl= Router URL. --etcdUrl= Etcd URL. --filepath= Directory to store functions in. --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 {} } diff --git a/fission.yaml b/fission.yaml index 5d26d128..785ed55e 100644 --- a/fission.yaml +++ b/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 diff --git a/fission/common.go b/fission/common.go index e2bbfec2..44eda243 100644 --- a/fission/common.go +++ b/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) } diff --git a/fission/main.go b/fission/main.go index adb7606c..3faa2b7f 100644 --- a/fission/main.go +++ b/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 { diff --git a/fission/watch.go b/fission/watch.go new file mode 100644 index 00000000..da08f861 --- /dev/null +++ b/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 to override.") + namespace = "default" + } + + objType := c.String("type") + if len(objType) == 0 { + fmt.Println("Object type unspecified, will watch pods. Use --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 +} diff --git a/kubewatcher/kubewatcher.go b/kubewatcher/kubewatcher.go new file mode 100644 index 00000000..0f87a6c9 --- /dev/null +++ b/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") + } +} diff --git a/kubewatcher/main.go b/kubewatcher/main.go new file mode 100644 index 00000000..92085d58 --- /dev/null +++ b/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 +} diff --git a/kubewatcher/publisher.go b/kubewatcher/publisher.go new file mode 100644 index 00000000..7d07e24f --- /dev/null +++ b/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) + } +) diff --git a/kubewatcher/watchSync.go b/kubewatcher/watchSync.go new file mode 100644 index 00000000..4eeee713 --- /dev/null +++ b/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) + } +} diff --git a/kubewatcher/webhookPublisher.go b/kubewatcher/webhookPublisher.go new file mode 100644 index 00000000..91dbb2d8 --- /dev/null +++ b/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 + } +} diff --git a/poolmgr/gp.go b/poolmgr/gp.go index f4ed46e8..11f2dd8a 100644 --- a/poolmgr/gp.go +++ b/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 diff --git a/resource.go b/resource.go index c5cb9dab..5695669f 100644 --- a/resource.go +++ b/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 +} diff --git a/router/httpTriggers.go b/router/httpTriggers.go index c9df3008..3e81a9ac 100644 --- a/router/httpTriggers.go +++ b/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) } } diff --git a/router/router.go b/router/router.go index 068b3ec9..da75dbe4 100644 --- a/router/router.go +++ b/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) { diff --git a/types.go b/types.go index f8325b4d..bfa67eab 100644 --- a/types.go +++ b/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 )