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, + } +}