From 0456df0d63593136395c8fa0f94fc511b5040845 Mon Sep 17 00:00:00 2001 From: Soam Vasani Date: Mon, 19 Dec 2016 14:38:44 -0800 Subject: [PATCH] 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.