From fa093829bce79963d5875d55816b9f6321d84c22 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 9 Dec 2016 23:22:18 -0800 Subject: [PATCH 01/27] Kubernetes events watcher Kubewatcher watches the Kubernetes API for Watches that have been configured on the fission controller. It then posts that event to the function configured in the watch. --- kubewatcher/kubewatcher.go | 199 +++++++++++++++++++++++++++++++++++ kubewatcher/main.go | 59 +++++++++++ kubewatcher/watchSync.go | 59 +++++++++++ kubewatcher/webhookposter.go | 79 ++++++++++++++ 4 files changed, 396 insertions(+) create mode 100644 kubewatcher/kubewatcher.go create mode 100644 kubewatcher/main.go create mode 100644 kubewatcher/watchSync.go create mode 100644 kubewatcher/webhookposter.go diff --git a/kubewatcher/kubewatcher.go b/kubewatcher/kubewatcher.go new file mode 100644 index 00000000..8a284dba --- /dev/null +++ b/kubewatcher/kubewatcher.go @@ -0,0 +1,199 @@ +/* +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" + + "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 ( + watchSubscription struct { + fission.Watch + kubeWatch watch.Interface + } + + kubeWatcherRequest struct { + requestType + watches []fission.Watch + responseChannel chan *kubeWatcherResponse + } + kubeWatcherResponse struct { + error + } + + KubeWatcher struct { + watches map[string]watchSubscription + kubernetesClient *kubernetes.Clientset + poster *Poster + requestChannel chan *kubeWatcherRequest + } +) + +func MakeKubeWatcher(kubernetesClient *kubernetes.Clientset, poster *Poster) *KubeWatcher { + kw := &KubeWatcher{ + watches: make(map[string]watchSubscription), + kubernetesClient: kubernetesClient, + poster: poster, + 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} + } + } +} + +// 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 + } + ws := &watchSubscription{ + Watch: *w, + kubeWatch: wi, + } + kw.watches[w.Metadata.Uid] = *ws + go ws.eventDispatchLoop(kw.poster) + 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)) + } + ws.kubeWatch.Stop() + delete(kw.watches, w.Metadata.Uid) + return nil +} + +func (ws *watchSubscription) eventDispatchLoop(poster *Poster) { + 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) + return + } + + var buf bytes.Buffer + err := printKubernetesObject(ev.Object, &buf) + if err != nil { + log.Println("Failed to serialize object: %v", err) + } + poster.Post(string(ev.Type), ws.Watch.Url, &buf) + } +} diff --git a/kubewatcher/main.go b/kubewatcher/main.go new file mode 100644 index 00000000..57418df1 --- /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 := MakePoster(routerUrl) + kubeWatch := MakeKubeWatcher(kubeClient, poster) + + client := client.MakeClient(controllerUrl) + MakeWatchSync(client, kubeWatch) + + return nil +} 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/webhookposter.go b/kubewatcher/webhookposter.go new file mode 100644 index 00000000..8fe5f48f --- /dev/null +++ b/kubewatcher/webhookposter.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 kubewatcher + +import ( + "io" + "log" + "net/http" + "strings" +) + +type ( + Poster struct { + routerUrl string + requestChannel chan *postRequest + } + postRequest struct { + eventType string + relativeUrl string + body io.Reader + } +) + +func MakePoster(routerUrl string) *Poster { + p := &Poster{ + routerUrl: strings.TrimSuffix(routerUrl, "/"), + requestChannel: make(chan *postRequest, 32), // buffered channel + } + go p.svc() + return p +} + +func (p *Poster) svc() { + for { + r := <-p.requestChannel + + url := p.routerUrl + r.relativeUrl + req, err := http.NewRequest("POST", url, r.body) + if err != nil { + log.Printf("Failed to create request to %v", r.relativeUrl) + } + req.Header.Add("X-Kubernetes-Event-Type", r.eventType) + req.Header.Add("X-Fission-Request-Async", "true") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + log.Printf("request failed: %v", r) + // TODO retries, persistence, etc. + } + + resp.Body.Close() + if resp.StatusCode != 200 { + log.Printf("request failed: %v", resp.StatusCode) + // TODO retries etc. + } + } +} + +func (p *Poster) Post(eventType, relativeUrl string, body io.Reader) { + p.requestChannel <- &postRequest{ + eventType: eventType, + relativeUrl: relativeUrl, + body: body, + } +} From e49af8ca78edf109a084d656a9f1466e2c850641 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 9 Dec 2016 23:25:29 -0800 Subject: [PATCH 02/27] Watch type Contains parameters to start a Kubernetes watch, and a function to call when the watch triggers. --- resource.go | 20 ++++++++++++++++++++ types.go | 15 +++++++++++++++ 2 files changed, 35 insertions(+) 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/types.go b/types.go index f8325b4d..044a5b00 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"` + + Url string `json:"url"` // POST request made to this URL. + } + // Errors returned by the Fission API. Error struct { Code errorCode `json:"code"` @@ -69,4 +83,5 @@ const ( ErrorNameExists ErrorInvalidArgument ErrorNoSpace + ErrorNotImplmented ) From 3a6d5bb0705a03229aa106650f0b6500abecffe3 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 9 Dec 2016 23:30:05 -0800 Subject: [PATCH 03/27] Add "internal" routes for all functions These routes allow us to call any function without explicitly defining a route for them. TODO: if we can serve these routes from a separate instance of router (controlled by a commandline flag or env var), we'd be able to keep this instance of the router "private" by not assigning it a loadbalancer/nodeport or other externally-visible service. --- common.go | 25 +++++++++++++++++++++++++ router/httpTriggers.go | 30 +++++++++++++++++++++++++++--- 2 files changed, 52 insertions(+), 3 deletions(-) create mode 100644 common.go diff --git a/common.go b/common.go new file mode 100644 index 00000000..51dcbb63 --- /dev/null +++ b/common.go @@ -0,0 +1,25 @@ +/* +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 { + return fmt.Sprintf("/fission-function/%v/%v", m.Name, m.Uid) +} diff --git a/router/httpTriggers.go b/router/httpTriggers.go index c9df3008..685efaf4 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 each function + for _, function := range ts.functions { + fh := &functionHandler{ + fmap: ts.functionServiceMap, + Function: function.Metadata, + poolmgr: ts.poolmgr, + } + muxRouter.HandleFunc(fission.UrlForFunction(&function.Metadata), + 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) } } From a69834157f96c3d1bd148db6cb546811af7e25e0 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 9 Dec 2016 23:42:03 -0800 Subject: [PATCH 04/27] Controller crud for Watch type Controller and client crud, API routes, and a unit test for the Watch type. --- controller/api.go | 17 +++++ controller/api_test.go | 43 +++++++++++-- controller/client/client.go | 87 ++++++++++++++++++++++++++ controller/watchApi.go | 120 ++++++++++++++++++++++++++++++++++++ controller/watchStore.go | 79 ++++++++++++++++++++++++ 5 files changed, 341 insertions(+), 5 deletions(-) create mode 100644 controller/watchApi.go create mode 100644 controller/watchStore.go 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..d61528ad 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: "", + }, + Url: "", + } + 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.Url = "" + 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..34e4a4d9 --- /dev/null +++ b/controller/watchApi.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 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.Url = 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 +} From 4ae0ab98e8d51412d70e06a342b0670bb037fc56 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 9 Dec 2016 23:43:27 -0800 Subject: [PATCH 05/27] Add kubewatcher to fission-bundle --- fission-bundle/main.go | 23 +++++++++++++++++------ 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/fission-bundle/main.go b/fission-bundle/main.go index d8c69fd9..59003a2b 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,12 +76,14 @@ 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'. @@ -92,6 +98,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 +115,9 @@ Options: runPoolmgr(port, controllerUrl, namespace) } + if arguments["--kubewatcher"] != nil { + runKubeWatcher(controllerUrl, routerUrl) + } + select {} } From 9dcd2b711f72c1c28425c62a314b0d62eb5cfc35 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 9 Dec 2016 23:58:31 -0800 Subject: [PATCH 06/27] Add kubewatcher deployment to yaml --- fission.yaml | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/fission.yaml b/fission.yaml index 5d26d128..e5abd265 100644 --- a/fission.yaml +++ b/fission.yaml @@ -22,7 +22,7 @@ metadata: spec: replicas: 1 template: - metadata: + metadata: labels: svc: controller spec: @@ -41,7 +41,7 @@ metadata: spec: replicas: 1 template: - metadata: + metadata: labels: svc: router spec: @@ -75,7 +75,7 @@ metadata: spec: replicas: 1 template: - metadata: + metadata: labels: svc: poolmgr spec: @@ -85,6 +85,25 @@ spec: command: ["/fission-bundle"] args: ["--poolmgrPort", "8888"] +--- +apiVersion: extensions/v1beta1 +kind: Deployment +metadata: + name: controller + namespace: fission +spec: + replicas: 1 + template: + metadata: + labels: + svc: controller + spec: + containers: + - name: controller + image: fission/fission-bundle:alpha + command: ["/fission-bundle"] + args: ["--kubewatcher"] + --- apiVersion: v1 kind: Service From 188280488c4322e0eb72789a5e3762a09c43419c Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 10 Dec 2016 19:53:13 -0800 Subject: [PATCH 07/27] Fix bug in populating watch url --- controller/watchApi.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/controller/watchApi.go b/controller/watchApi.go index 34e4a4d9..329b833b 100644 --- a/controller/watchApi.go +++ b/controller/watchApi.go @@ -55,7 +55,14 @@ func (api *API) WatchApiCreate(w http.ResponseWriter, r *http.Request) { api.respondWithError(w, err) return } - watch.Url = fission.UrlForFunction(&watch.Function) + + function, err := api.FunctionStore.Get(&watch.Function) + if err != nil { + api.respondWithError(w, err) + return + } + + watch.Url = fission.UrlForFunction(&function.Metadata) uid, err := api.WatchStore.Create(&watch) if err != nil { From ddb189bda20f14423c62ceaad94ac796cbb95183 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 10 Dec 2016 19:53:46 -0800 Subject: [PATCH 08/27] Kube requests: add content-type --- kubewatcher/webhookposter.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/kubewatcher/webhookposter.go b/kubewatcher/webhookposter.go index 8fe5f48f..972be4c8 100644 --- a/kubewatcher/webhookposter.go +++ b/kubewatcher/webhookposter.go @@ -49,12 +49,13 @@ func (p *Poster) svc() { r := <-p.requestChannel url := p.routerUrl + r.relativeUrl + log.Printf("Making request to %v", url) req, err := http.NewRequest("POST", url, r.body) if err != nil { log.Printf("Failed to create request to %v", r.relativeUrl) } + req.Header.Add("Content-Type", "application/json") req.Header.Add("X-Kubernetes-Event-Type", r.eventType) - req.Header.Add("X-Fission-Request-Async", "true") resp, err := http.DefaultClient.Do(req) if err != nil { From f29e69c5819784a64462b95a058572febd9701d7 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 10 Dec 2016 19:55:20 -0800 Subject: [PATCH 09/27] Fix cleanup on duplicate function specialization --- poolmgr/gp.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 From 2379cb6f814e451c6f5759bdb35e8a462fb63537 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 10 Dec 2016 19:55:43 -0800 Subject: [PATCH 10/27] Add watches to fission CLI --- fission/main.go | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) 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 { From 7918b89fd2d0868b3f73b8181b8e49efb01faa36 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 10 Dec 2016 19:56:15 -0800 Subject: [PATCH 11/27] Bugfix in kubewatcher deployment yaml --- fission.yaml | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/fission.yaml b/fission.yaml index e5abd265..785ed55e 100644 --- a/fission.yaml +++ b/fission.yaml @@ -28,7 +28,7 @@ spec: spec: containers: - name: controller - image: fission/fission-bundle:alpha + image: fission/fission-bundle:alpha10 command: ["/fission-bundle"] args: ["--controllerPort", "8888", "--filepath", "/filestore"] @@ -47,7 +47,7 @@ spec: spec: containers: - name: router - image: fission/fission-bundle:alpha + image: fission/fission-bundle:alpha10 command: ["/fission-bundle"] args: ["--routerPort", "8888"] @@ -81,7 +81,7 @@ spec: spec: containers: - name: poolmgr - image: fission/fission-bundle:alpha + image: fission/fission-bundle:alpha10 command: ["/fission-bundle"] args: ["--poolmgrPort", "8888"] @@ -89,18 +89,18 @@ spec: apiVersion: extensions/v1beta1 kind: Deployment metadata: - name: controller + name: kubewatcher namespace: fission spec: replicas: 1 template: metadata: labels: - svc: controller + svc: kubewatcher spec: containers: - - name: controller - image: fission/fission-bundle:alpha + - name: kubewatcher + image: fission/fission-bundle:alpha10 command: ["/fission-bundle"] args: ["--kubewatcher"] From 193cf594beafea531d01d7931f1cce99314780f0 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sat, 10 Dec 2016 19:56:45 -0800 Subject: [PATCH 12/27] Let fission CLI tolerate URL with/without leading http:// --- fission/common.go | 3 +++ 1 file changed, 3 insertions(+) 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) } From 0fb4cc1536b08d19403a80c8be503e85940b3558 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sun, 11 Dec 2016 00:07:25 -0800 Subject: [PATCH 13/27] Use latest version of function for watches --- common.go | 7 ++++++- controller/watchApi.go | 8 +------- router/httpTriggers.go | 2 +- 3 files changed, 8 insertions(+), 9 deletions(-) diff --git a/common.go b/common.go index 51dcbb63..5f9a2c56 100644 --- a/common.go +++ b/common.go @@ -21,5 +21,10 @@ import ( ) func UrlForFunction(m *Metadata) string { - return fmt.Sprintf("/fission-function/%v/%v", m.Name, m.Uid) + 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/watchApi.go b/controller/watchApi.go index 329b833b..b1c67d69 100644 --- a/controller/watchApi.go +++ b/controller/watchApi.go @@ -56,13 +56,7 @@ func (api *API) WatchApiCreate(w http.ResponseWriter, r *http.Request) { return } - function, err := api.FunctionStore.Get(&watch.Function) - if err != nil { - api.respondWithError(w, err) - return - } - - watch.Url = fission.UrlForFunction(&function.Metadata) + watch.Url = fission.UrlForFunction(&watch.Function) uid, err := api.WatchStore.Create(&watch) if err != nil { diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 685efaf4..af8211d1 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -71,7 +71,7 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { for _, function := range ts.functions { fh := &functionHandler{ fmap: ts.functionServiceMap, - Function: function.Metadata, + Function: fission.Metadata{Name: function.Metadata.Name}, poolmgr: ts.poolmgr, } muxRouter.HandleFunc(fission.UrlForFunction(&function.Metadata), From 5e3adc0b5953da4804999e3a06ec7232adbbaae2 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sun, 11 Dec 2016 19:58:04 -0800 Subject: [PATCH 14/27] Log on function exception --- environments/nodejs/server.js | 1 + 1 file changed, 1 insertion(+) 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") } }); From f1e1da4ceb80e2040337891d8c0c1f3c693bef8e Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sun, 11 Dec 2016 19:58:26 -0800 Subject: [PATCH 15/27] Bugfix in fission-bundle arg parsing --- fission-bundle/main.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/fission-bundle/main.go b/fission-bundle/main.go index 59003a2b..26e06d97 100644 --- a/fission-bundle/main.go +++ b/fission-bundle/main.go @@ -87,6 +87,7 @@ Options: --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 { @@ -115,7 +116,7 @@ Options: runPoolmgr(port, controllerUrl, namespace) } - if arguments["--kubewatcher"] != nil { + if arguments["--kubewatcher"] == true { runKubeWatcher(controllerUrl, routerUrl) } From b7906905c2d72e6117fe8f0703c86b4292d99d6d Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Sun, 11 Dec 2016 19:59:10 -0800 Subject: [PATCH 16/27] Make the internal triggers point to the latest version --- router/httpTriggers.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/router/httpTriggers.go b/router/httpTriggers.go index af8211d1..3e81a9ac 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -67,15 +67,15 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { muxRouter.HandleFunc(trigger.UrlPattern, fh.handler) } - // Internal triggers for each function + // 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: fission.Metadata{Name: function.Metadata.Name}, + Function: m, poolmgr: ts.poolmgr, } - muxRouter.HandleFunc(fission.UrlForFunction(&function.Metadata), - fh.handler) + muxRouter.HandleFunc(fission.UrlForFunction(&m), fh.handler) } return muxRouter From 5c6dcca19d7548e92894a302621c71a5d7a6d033 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Mon, 12 Dec 2016 08:46:02 -0800 Subject: [PATCH 17/27] Pass obj type to functions --- kubewatcher/kubewatcher.go | 2 +- kubewatcher/webhookposter.go | 12 ++++++++++-- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/kubewatcher/kubewatcher.go b/kubewatcher/kubewatcher.go index 8a284dba..6c7a9c50 100644 --- a/kubewatcher/kubewatcher.go +++ b/kubewatcher/kubewatcher.go @@ -194,6 +194,6 @@ func (ws *watchSubscription) eventDispatchLoop(poster *Poster) { if err != nil { log.Println("Failed to serialize object: %v", err) } - poster.Post(string(ev.Type), ws.Watch.Url, &buf) + poster.Post(string(ev.Type), ws.Watch.ObjType, ws.Watch.Url, &buf) } } diff --git a/kubewatcher/webhookposter.go b/kubewatcher/webhookposter.go index 972be4c8..95fe5246 100644 --- a/kubewatcher/webhookposter.go +++ b/kubewatcher/webhookposter.go @@ -18,6 +18,7 @@ package kubewatcher import ( "io" + "io/ioutil" "log" "net/http" "strings" @@ -30,6 +31,7 @@ type ( } postRequest struct { eventType string + objType string relativeUrl string body io.Reader } @@ -56,6 +58,7 @@ func (p *Poster) svc() { } req.Header.Add("Content-Type", "application/json") req.Header.Add("X-Kubernetes-Event-Type", r.eventType) + req.Header.Add("X-Kubernetes-Object-Type", r.objType) resp, err := http.DefaultClient.Do(req) if err != nil { @@ -63,17 +66,22 @@ func (p *Poster) svc() { // TODO retries, persistence, etc. } - resp.Body.Close() if resp.StatusCode != 200 { log.Printf("request failed: %v", resp.StatusCode) + body, err := ioutil.ReadAll(resp.Body) + if err == nil { + log.Printf("request error: %v", body) + } // TODO retries etc. } + resp.Body.Close() } } -func (p *Poster) Post(eventType, relativeUrl string, body io.Reader) { +func (p *Poster) Post(eventType, objType, relativeUrl string, body io.Reader) { p.requestChannel <- &postRequest{ eventType: eventType, + objType: objType, relativeUrl: relativeUrl, body: body, } From 4bc7480acdaf1dab6180738ca225022c1a98b7e0 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 15 Dec 2016 09:33:02 -0800 Subject: [PATCH 18/27] Add request logging to router --- router/router.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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) { From 53c49d148537ad2b261ba5ee656f93d7d6e82f51 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 15 Dec 2016 09:34:03 -0800 Subject: [PATCH 19/27] Add an atomic bool to track if a watch is stopped Saw a watch channel close once without actually calling watch.Stop. Need to investigate this a bit and handle it if necessary. --- kubewatcher/kubewatcher.go | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/kubewatcher/kubewatcher.go b/kubewatcher/kubewatcher.go index 6c7a9c50..db403974 100644 --- a/kubewatcher/kubewatcher.go +++ b/kubewatcher/kubewatcher.go @@ -24,6 +24,7 @@ import ( "io" "log" "strings" + "sync/atomic" "k8s.io/client-go/1.5/kubernetes" "k8s.io/client-go/1.5/pkg/api" @@ -43,6 +44,7 @@ type ( watchSubscription struct { fission.Watch kubeWatch watch.Interface + stopped *int32 } kubeWatcherRequest struct { @@ -159,9 +161,11 @@ func (kw *KubeWatcher) addWatch(w *fission.Watch) error { 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.poster) @@ -175,8 +179,9 @@ func (kw *KubeWatcher) removeWatch(w *fission.Watch) error { return fission.MakeError(fission.ErrorNotFound, fmt.Sprintf("watch doesn't exist: %v", w.Metadata)) } - ws.kubeWatch.Stop() delete(kw.watches, w.Metadata.Uid) + atomic.StoreInt32(ws.stopped, 1) + ws.kubeWatch.Stop() return nil } @@ -186,7 +191,7 @@ func (ws *watchSubscription) eventDispatchLoop(poster *Poster) { ev, more := <-ws.kubeWatch.ResultChan() if !more { log.Println("Watch stopped", ws.Watch.Metadata.Name) - return + break } var buf bytes.Buffer @@ -196,4 +201,8 @@ func (ws *watchSubscription) eventDispatchLoop(poster *Poster) { } poster.Post(string(ev.Type), ws.Watch.ObjType, ws.Watch.Url, &buf) } + if !atomic.LoadInt32(ws.stopped) { + // TODO re-watch. What about resource version? + log.Panicf("Watch channel closed unexpectedly") + } } From 215f4fc30fb4795376357f2fe70870f2193d8ba1 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Thu, 15 Dec 2016 19:48:45 -0800 Subject: [PATCH 20/27] comments --- kubewatcher/kubewatcher.go | 4 ++++ kubewatcher/webhookposter.go | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/kubewatcher/kubewatcher.go b/kubewatcher/kubewatcher.go index db403974..baf48d2e 100644 --- a/kubewatcher/kubewatcher.go +++ b/kubewatcher/kubewatcher.go @@ -199,6 +199,10 @@ func (ws *watchSubscription) eventDispatchLoop(poster *Poster) { if err != nil { log.Println("Failed to serialize object: %v", err) } + // TODO re: objtype -- mabye we should use runtime + // type info to get the obj type, in case the watch + // can send multiple types of objects (it probably + // sends a different type when ev.Type == ERROR?) poster.Post(string(ev.Type), ws.Watch.ObjType, ws.Watch.Url, &buf) } if !atomic.LoadInt32(ws.stopped) { diff --git a/kubewatcher/webhookposter.go b/kubewatcher/webhookposter.go index 95fe5246..e4a19e71 100644 --- a/kubewatcher/webhookposter.go +++ b/kubewatcher/webhookposter.go @@ -72,7 +72,7 @@ func (p *Poster) svc() { if err == nil { log.Printf("request error: %v", body) } - // TODO retries etc. + // TODO retries, persistence, etc. } resp.Body.Close() } From de30886c9e65482b1940f583cced1cbb105f7ea1 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 16 Dec 2016 00:43:45 -0800 Subject: [PATCH 21/27] compile error --- kubewatcher/kubewatcher.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/kubewatcher/kubewatcher.go b/kubewatcher/kubewatcher.go index baf48d2e..be6def8e 100644 --- a/kubewatcher/kubewatcher.go +++ b/kubewatcher/kubewatcher.go @@ -205,7 +205,7 @@ func (ws *watchSubscription) eventDispatchLoop(poster *Poster) { // sends a different type when ev.Type == ERROR?) poster.Post(string(ev.Type), ws.Watch.ObjType, ws.Watch.Url, &buf) } - if !atomic.LoadInt32(ws.stopped) { + if atomic.LoadInt32(ws.stopped) != 0 { // TODO re-watch. What about resource version? log.Panicf("Watch channel closed unexpectedly") } From dddb50c00f09722e835f7d1bee8d125d2e67eb30 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Fri, 16 Dec 2016 00:54:06 -0800 Subject: [PATCH 22/27] cli watch impl --- fission/watch.go | 120 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 fission/watch.go 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 +} From 0456df0d63593136395c8fa0f94fc511b5040845 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Mon, 19 Dec 2016 14:38:44 -0800 Subject: [PATCH 23/27] Refactor kubewatcher/webhook publisher Create a Publisher interface. Refactor webhook publisher to satisfy that interface. Add some limited retries to webhook publisher. --- common.go | 4 +- kubewatcher/kubewatcher.go | 39 ++++------ kubewatcher/main.go | 2 +- kubewatcher/publisher.go | 30 ++++++++ kubewatcher/webhookPublisher.go | 131 ++++++++++++++++++++++++++++++++ kubewatcher/webhookposter.go | 88 --------------------- types.go | 2 +- 7 files changed, 181 insertions(+), 115 deletions(-) create mode 100644 kubewatcher/publisher.go create mode 100644 kubewatcher/webhookPublisher.go delete mode 100644 kubewatcher/webhookposter.go diff --git a/common.go b/common.go index 5f9a2c56..15b22744 100644 --- a/common.go +++ b/common.go @@ -21,7 +21,9 @@ import ( ) func UrlForFunction(m *Metadata) string { - prefix := "/fission-function" + // TODO this assumes the router's namespace is the same as whatever is hitting + // this url -- so e.g. kubewatcher will have to run in the same ns as router. + prefix := "http://router/fission-function" if len(m.Uid) > 0 { return fmt.Sprintf("%v/%v/%v", prefix, m.Name, m.Uid) } else { diff --git a/kubewatcher/kubewatcher.go b/kubewatcher/kubewatcher.go index be6def8e..0f87a6c9 100644 --- a/kubewatcher/kubewatcher.go +++ b/kubewatcher/kubewatcher.go @@ -41,6 +41,13 @@ const ( ) type ( + KubeWatcher struct { + watches map[string]watchSubscription + kubernetesClient *kubernetes.Clientset + requestChannel chan *kubeWatcherRequest + publisher Publisher + } + watchSubscription struct { fission.Watch kubeWatch watch.Interface @@ -55,20 +62,13 @@ type ( kubeWatcherResponse struct { error } - - KubeWatcher struct { - watches map[string]watchSubscription - kubernetesClient *kubernetes.Clientset - poster *Poster - requestChannel chan *kubeWatcherRequest - } ) -func MakeKubeWatcher(kubernetesClient *kubernetes.Clientset, poster *Poster) *KubeWatcher { +func MakeKubeWatcher(kubernetesClient *kubernetes.Clientset, publisher Publisher) *KubeWatcher { kw := &KubeWatcher{ watches: make(map[string]watchSubscription), kubernetesClient: kubernetesClient, - poster: poster, + publisher: publisher, requestChannel: make(chan *kubeWatcherRequest), } go kw.svc() @@ -112,7 +112,7 @@ func (kw *KubeWatcher) svc() { } } -// lifted from kubernetes/pkg/kubectl/resource_printer.go +// 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: @@ -168,7 +168,7 @@ func (kw *KubeWatcher) addWatch(w *fission.Watch) error { stopped: &stopped, } kw.watches[w.Metadata.Uid] = *ws - go ws.eventDispatchLoop(kw.poster) + go ws.eventDispatchLoop(kw.publisher) return nil } @@ -185,7 +185,7 @@ func (kw *KubeWatcher) removeWatch(w *fission.Watch) error { return nil } -func (ws *watchSubscription) eventDispatchLoop(poster *Poster) { +func (ws *watchSubscription) eventDispatchLoop(publisher Publisher) { log.Println("Listening to watch ", ws.Watch.Metadata.Name) for { ev, more := <-ws.kubeWatch.ResultChan() @@ -193,20 +193,11 @@ func (ws *watchSubscription) eventDispatchLoop(poster *Poster) { log.Println("Watch stopped", ws.Watch.Metadata.Name) break } - - var buf bytes.Buffer - err := printKubernetesObject(ev.Object, &buf) - if err != nil { - log.Println("Failed to serialize object: %v", err) - } - // TODO re: objtype -- mabye we should use runtime - // type info to get the obj type, in case the watch - // can send multiple types of objects (it probably - // sends a different type when ev.Type == ERROR?) - poster.Post(string(ev.Type), ws.Watch.ObjType, ws.Watch.Url, &buf) + publisher.Publish(ev, ws.Watch.Target) } if atomic.LoadInt32(ws.stopped) != 0 { - // TODO re-watch. What about resource version? + // 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 index 57418df1..85837277 100644 --- a/kubewatcher/main.go +++ b/kubewatcher/main.go @@ -49,7 +49,7 @@ func Start(controllerUrl string, routerUrl string) error { if err != nil { return err } - poster := MakePoster(routerUrl) + poster := MakeWebhookPublisher() kubeWatch := MakeKubeWatcher(kubeClient, poster) client := client.MakeClient(controllerUrl) 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/webhookPublisher.go b/kubewatcher/webhookPublisher.go new file mode 100644 index 00000000..8d910ca5 --- /dev/null +++ b/kubewatcher/webhookPublisher.go @@ -0,0 +1,131 @@ +/* +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" + "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 + } + publishRequest struct { + url string + watchEvent watch.Event + retries int + retryDelay time.Duration + } +) + +// The caller must make one of these per URL +func MakeWebhookPublisher() *WebhookPublisher { + p := &WebhookPublisher{ + 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) { + + log.Printf("Making HTTP request to %v", r.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", r.url, &buf) + if err != nil { + log.Printf("Failed to create request to %v", r.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).Name()) + + // Make the request + resp, err := http.DefaultClient.Do(req) + if err == nil && resp.StatusCode == 200 { + resp.Body.Close() + // all done. + 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", body) + } + } + + // Schedule a retry, or give up + 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", r.url) + // Event dropped + } +} diff --git a/kubewatcher/webhookposter.go b/kubewatcher/webhookposter.go deleted file mode 100644 index e4a19e71..00000000 --- a/kubewatcher/webhookposter.go +++ /dev/null @@ -1,88 +0,0 @@ -/* -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 ( - "io" - "io/ioutil" - "log" - "net/http" - "strings" -) - -type ( - Poster struct { - routerUrl string - requestChannel chan *postRequest - } - postRequest struct { - eventType string - objType string - relativeUrl string - body io.Reader - } -) - -func MakePoster(routerUrl string) *Poster { - p := &Poster{ - routerUrl: strings.TrimSuffix(routerUrl, "/"), - requestChannel: make(chan *postRequest, 32), // buffered channel - } - go p.svc() - return p -} - -func (p *Poster) svc() { - for { - r := <-p.requestChannel - - url := p.routerUrl + r.relativeUrl - log.Printf("Making request to %v", url) - req, err := http.NewRequest("POST", url, r.body) - if err != nil { - log.Printf("Failed to create request to %v", r.relativeUrl) - } - req.Header.Add("Content-Type", "application/json") - req.Header.Add("X-Kubernetes-Event-Type", r.eventType) - req.Header.Add("X-Kubernetes-Object-Type", r.objType) - - resp, err := http.DefaultClient.Do(req) - if err != nil { - log.Printf("request failed: %v", r) - // TODO retries, persistence, etc. - } - - if resp.StatusCode != 200 { - log.Printf("request failed: %v", resp.StatusCode) - body, err := ioutil.ReadAll(resp.Body) - if err == nil { - log.Printf("request error: %v", body) - } - // TODO retries, persistence, etc. - } - resp.Body.Close() - } -} - -func (p *Poster) Post(eventType, objType, relativeUrl string, body io.Reader) { - p.requestChannel <- &postRequest{ - eventType: eventType, - objType: objType, - relativeUrl: relativeUrl, - body: body, - } -} diff --git a/types.go b/types.go index 044a5b00..bfa67eab 100644 --- a/types.go +++ b/types.go @@ -64,7 +64,7 @@ type ( Function Metadata `json:"function"` - Url string `json:"url"` // POST request made to this URL. + Target string `json:"target"` // Watch publish target (URL, NATS stream, etc) } // Errors returned by the Fission API. From 4d75c3fa4eea1087574072efa9edadb3c3daa390 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 20 Dec 2016 12:11:22 -0800 Subject: [PATCH 24/27] Make internal urls relative (as before) --- common.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/common.go b/common.go index 15b22744..5f9a2c56 100644 --- a/common.go +++ b/common.go @@ -21,9 +21,7 @@ import ( ) func UrlForFunction(m *Metadata) string { - // TODO this assumes the router's namespace is the same as whatever is hitting - // this url -- so e.g. kubewatcher will have to run in the same ns as router. - prefix := "http://router/fission-function" + prefix := "/fission-function" if len(m.Uid) > 0 { return fmt.Sprintf("%v/%v/%v", prefix, m.Name, m.Uid) } else { From 8704761fd0dac39a1c5fb1168457aa5a54d322a8 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 20 Dec 2016 12:11:54 -0800 Subject: [PATCH 25/27] watch.Url renamed to watch.Target --- controller/api_test.go | 4 ++-- controller/watchApi.go | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/controller/api_test.go b/controller/api_test.go index d61528ad..f241866d 100644 --- a/controller/api_test.go +++ b/controller/api_test.go @@ -199,7 +199,7 @@ func TestWatchApi(t *testing.T) { Name: "foo", Uid: "", }, - Url: "", + Target: "", } m, err := g.client.WatchCreate(testWatch) panicIf(err) @@ -208,7 +208,7 @@ func TestWatchApi(t *testing.T) { w, err := g.client.WatchGet(m) panicIf(err) testWatch.Metadata.Uid = m.Uid - w.Url = "" + w.Target = "" assert(*testWatch == *w, "watch should match after reading") testWatch.Metadata.Name = "yyy" diff --git a/controller/watchApi.go b/controller/watchApi.go index b1c67d69..cbec5e6b 100644 --- a/controller/watchApi.go +++ b/controller/watchApi.go @@ -56,7 +56,7 @@ func (api *API) WatchApiCreate(w http.ResponseWriter, r *http.Request) { return } - watch.Url = fission.UrlForFunction(&watch.Function) + watch.Target = fission.UrlForFunction(&watch.Function) uid, err := api.WatchStore.Create(&watch) if err != nil { From c5425180e8479267bbd279ead845124be56d7d9e Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 20 Dec 2016 12:19:51 -0800 Subject: [PATCH 26/27] Make webhook publisher use relative urls --- kubewatcher/main.go | 2 +- kubewatcher/webhookPublisher.go | 19 +++++++++++-------- 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/kubewatcher/main.go b/kubewatcher/main.go index 85837277..92085d58 100644 --- a/kubewatcher/main.go +++ b/kubewatcher/main.go @@ -49,7 +49,7 @@ func Start(controllerUrl string, routerUrl string) error { if err != nil { return err } - poster := MakeWebhookPublisher() + poster := MakeWebhookPublisher(routerUrl) kubeWatch := MakeKubeWatcher(kubeClient, poster) client := client.MakeClient(controllerUrl) diff --git a/kubewatcher/webhookPublisher.go b/kubewatcher/webhookPublisher.go index 8d910ca5..1440162e 100644 --- a/kubewatcher/webhookPublisher.go +++ b/kubewatcher/webhookPublisher.go @@ -22,6 +22,7 @@ import ( "log" "net/http" "reflect" + "strings" "time" "k8s.io/client-go/1.5/pkg/watch" @@ -34,6 +35,8 @@ type ( maxRetries int retryDelay time.Duration + + baseUrl string } publishRequest struct { url string @@ -43,11 +46,10 @@ type ( } ) -// The caller must make one of these per URL -func MakeWebhookPublisher() *WebhookPublisher { +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, @@ -74,7 +76,8 @@ func (p *WebhookPublisher) svc() { func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) { - log.Printf("Making HTTP request to %v", r.url) + url := p.baseUrl + "/" + strings.TrimPrefix(r.url, "/") + log.Printf("Making HTTP request to %v", url) // Serialize the object var buf bytes.Buffer @@ -85,9 +88,9 @@ func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) { } // Create request - req, err := http.NewRequest("POST", r.url, &buf) + req, err := http.NewRequest("POST", url, &buf) if err != nil { - log.Printf("Failed to create request to %v", r.url) + log.Printf("Failed to create request to %v", url) // can't do anything more, drop the event. return } @@ -113,7 +116,7 @@ func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) { body, err := ioutil.ReadAll(resp.Body) resp.Body.Close() if err == nil { - log.Printf("request error: %v", body) + log.Printf("request error: %v", string(body)) } } @@ -125,7 +128,7 @@ func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) { p.requestChannel <- r }) } else { - log.Printf("Final retry failed, giving up on %v", r.url) + log.Printf("Final retry failed, giving up on %v", url) // Event dropped } } From d721df998d638e79bbb0d74ff4d807ffb8e44f26 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Tue, 20 Dec 2016 14:16:33 -0800 Subject: [PATCH 27/27] Fix object-type header passed to event handlers --- kubewatcher/webhookPublisher.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/kubewatcher/webhookPublisher.go b/kubewatcher/webhookPublisher.go index 1440162e..91dbb2d8 100644 --- a/kubewatcher/webhookPublisher.go +++ b/kubewatcher/webhookPublisher.go @@ -98,13 +98,14 @@ func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) { // 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).Name()) + 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() - // all done. return } @@ -120,7 +121,7 @@ func (p *WebhookPublisher) makeHttpRequest(r *publishRequest) { } } - // Schedule a retry, or give up + // Schedule a retry, or give up if out of retries r.retries-- if r.retries > 0 { r.retryDelay *= time.Duration(2)