Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor kubewatcher/webhook publisher

Browse Source 
Create a Publisher interface. Refactor webhook publisher to satisfy
that interface.

Add some limited retries to webhook publisher.
This commit is contained in:
Soam Vasani
2016-12-19 14:38:44 -08:00
parent dddb50c00f
commit 0456df0d63
7 changed files with 181 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common.go
+3 -1
View File
@@ -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 {
kubewatcher/kubewatcher.go
+15 -24
View File
@@ -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")
}
}
kubewatcher/main.go
+1 -1
View File
@@ -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)
kubewatcher/publisher.go
+30
View File
@@ -0,0 +1,30 @@
/*
Copyright 2016 The Fission Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package kubewatcher
import (
"k8s.io/client-go/1.5/pkg/watch"
)
type (
Publisher interface {
// Publish an event to a "target". Target's meaning depends on the
// publisher: it's a URL in the case of a webhook publisher, or a queue
// name in a queue-based publisher such as NATS.
Publish(event watch.Event, target string)
}
)
kubewatcher/webhookPublisher.go
+131
View File
@@ -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
}
}
kubewatcher/webhookposter.go
-88
View File
@@ -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,
}
}
types.go
+1 -1
View File
@@ -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.