From 1939278d58b2d0d91590f221a555921c06066802 Mon Sep 17 00:00:00 2001 From: Darren Shepherd Date: Sat, 20 Feb 2016 23:17:01 -0700 Subject: [PATCH 1/3] Propagate all events to Rancher --- kubernetesevents/change_handler.go | 37 ++++++++++++++++++++++++++++++ main.go | 10 ++++++++ 2 files changed, 47 insertions(+) create mode 100644 kubernetesevents/change_handler.go diff --git a/kubernetesevents/change_handler.go b/kubernetesevents/change_handler.go new file mode 100644 index 0000000..ad7ebbc --- /dev/null +++ b/kubernetesevents/change_handler.go @@ -0,0 +1,37 @@ +package kubernetesevents + +import ( + "github.com/rancher/go-rancher/client" + "github.com/rancher/kubernetes-agent/kubernetesclient" + "github.com/rancher/kubernetes-model/model" +) + +func NewChangeHandler(rancherClient *client.RancherClient, kubernetesClient *kubernetesclient.Client, kindHandled string) *ChangeHandler { + return &ChangeHandler{ + rancherClient: rancherClient, + kClient: kubernetesClient, + kindHandled: kindHandled, + } +} + +type ChangeHandler struct { + rancherClient *client.RancherClient + kClient *kubernetesclient.Client + kindHandled string +} + +func (h *ChangeHandler) GetKindHandled() string { + return h.kindHandled +} + +func (h *ChangeHandler) Handle(event model.WatchEvent) error { + _, err := h.rancherClient.Publish.Create(&client.Publish{ + Name: "service.kubernetes.change", + Data: map[string]interface{}{ + "type": event.Type, + "object": event.Object, + }, + }) + + return err +} diff --git a/main.go b/main.go index 06d4c98..2ff3c6f 100644 --- a/main.go +++ b/main.go @@ -53,6 +53,11 @@ func main() { Usage: "Port to configure an HTTP health check listener on", EnvVar: "HEALTH_CHECK_PORT", }, + cli.StringSliceFlag{ + Name: "watch-kind", + Value: &cli.StringSlice{"namespaces", "services", "replicationcontrollers", "pods"}, + Usage: "Which k8s kinds to watch and report changes to Rancher", + }, } app.Run(os.Args) @@ -73,6 +78,11 @@ func launch(c *cli.Context) { svcHandler := kubernetesevents.NewHandler(rClient, kClient, kubernetesevents.ServiceKind) handlers := []kubernetesevents.Handler{svcHandler} + log.Info("Watching changes for kinds: ", c.StringSlice("watch-kind")) + for _, kind := range c.StringSlice("watch-kind") { + handlers = append(handlers, kubernetesevents.NewChangeHandler(rClient, kClient, kind)) + } + go func(rc chan error) { err := kubernetesevents.ConnectToEventStream(handlers, conf) log.Errorf("Kubernetes stream listener exited with error: %s", err) From c2ce8a089c6958d3324a7c728056f5b3f3c66c64 Mon Sep 17 00:00:00 2001 From: Antmanler Date: Fri, 4 Dec 2015 19:17:09 +0800 Subject: [PATCH 2/3] Populate labels for k8s greater than 1.1.0 --- .../eventhandlers/provide_lables_handler.go | 67 ++++++++++--------- 1 file changed, 36 insertions(+), 31 deletions(-) diff --git a/rancherevents/eventhandlers/provide_lables_handler.go b/rancherevents/eventhandlers/provide_lables_handler.go index 9babcb5..271f280 100644 --- a/rancherevents/eventhandlers/provide_lables_handler.go +++ b/rancherevents/eventhandlers/provide_lables_handler.go @@ -1,9 +1,10 @@ package eventhandlers import ( - log "github.com/Sirupsen/logrus" "strings" + log "github.com/Sirupsen/logrus" + "github.com/mitchellh/mapstructure" revents "github.com/rancher/go-machine-service/events" "github.com/rancher/go-rancher/client" "github.com/rancher/kubernetes-agent/kubernetesclient" @@ -85,37 +86,41 @@ func (h *syncHandler) Handler(event *revents.Event, cli *client.RancherClient) e return nil } -func (h *syncHandler) getPod(event *revents.Event) (string, string) { - // TODO Rewrite this horror - data := event.Data - if ihm, ok := data["instanceHostMap"]; ok { - if ihmMap, ok := ihm.(map[string]interface{}); ok { - if i, ok := ihmMap["instance"]; ok { - if iMap, ok := i.(map[string]interface{}); ok { - if d, ok := iMap["data"]; ok { - if dMap, ok := d.(map[string]interface{}); ok { - if f, ok := dMap["fields"]; ok { - if fMap, ok := f.(map[string]interface{}); ok { - if labels, ok := fMap["labels"]; ok { - if lMap, ok := labels.(map[string]interface{}); ok { - if l, ok := lMap["io.kubernetes.pod.name"]; ok { - if label, ok := l.(string); ok { - parts := strings.SplitN(label, "/", 2) - if len(parts) == 2 { - return parts[0], parts[1] - } - } - } - } - } - } - } - } - } - } - } +func (h *syncHandler) getPod(event *revents.Event) (ns, name string) { + ihm := &struct { + IHM struct { + I struct { + D struct { + F struct { + Labels map[string]string `mapstructure:"labels"` + } `mapstructure:"fields"` + } `mapstructure:"data"` + } `mapstructure:"instance"` + } `mapstructure:"instanceHostMap"` + }{} + + err := mapstructure.Decode(event.Data, &ihm) + if err != nil { + log.Error("Cannot parse event") + return + } + + labels := ihm.IHM.I.D.F.Labels + if len(labels) == 0 { + return + } + + var ok bool + if ns, ok = labels["io.kubernetes.pod.namespace"]; ok { + // version >= 1.2 + name = labels["io.kubernetes.pod.name"] + } else if name, ok = labels["io.kubernetes.pod.name"]; ok { + // try to parse + parts := strings.SplitN(name, "/", 2) + if len(parts) == 0 { + ns, name = parts[0], parts[1] } } - return "", "" + return } From 0404f3d84a06562995637bca843740eb93337dc3 Mon Sep 17 00:00:00 2001 From: Darren Shepherd Date: Thu, 25 Feb 2016 21:25:06 -0700 Subject: [PATCH 3/3] Small fix --- rancherevents/eventhandlers/provide_lables_handler.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rancherevents/eventhandlers/provide_lables_handler.go b/rancherevents/eventhandlers/provide_lables_handler.go index 271f280..eba1399 100644 --- a/rancherevents/eventhandlers/provide_lables_handler.go +++ b/rancherevents/eventhandlers/provide_lables_handler.go @@ -117,7 +117,7 @@ func (h *syncHandler) getPod(event *revents.Event) (ns, name string) { } else if name, ok = labels["io.kubernetes.pod.name"]; ok { // try to parse parts := strings.SplitN(name, "/", 2) - if len(parts) == 0 { + if len(parts) == 2 { ns, name = parts[0], parts[1] } }