baalajimaestro/traefik
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

Fix Kubernetes watch SSL

Browse source 
Signed-off-by: Emile Vauge <emile@vauge.com>
This commit is contained in:
Emile Vauge 2016-04-26 22:13:45 +02:00
parent 9053f9bb98
commit 4ff4e4e626
No known key found for this signature in database
GPG key ID: D808B4C167352E59
1 changed files with 21 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
provider/k8s/client.go
View file

@ -7,11 +7,9 @@ import (
"fmt" "fmt"
"github.com/containous/traefik/safe" "github.com/containous/traefik/safe"
"github.com/parnurzeal/gorequest" "github.com/parnurzeal/gorequest"
"net"
"net/http" "net/http"
"net/url" "net/url"
"strings" "strings"
"time"
) )
const ( const (
@ -57,7 +55,7 @@ func (c *clientImpl) GetIngresses(predicate func(Ingress) bool) ([]Ingress, erro
body, err := c.do(c.request(getURL)) body, err := c.do(c.request(getURL))
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to create request: GET %q : %v", getURL, err) return nil, fmt.Errorf("failed to create ingresses request: GET %q : %v", getURL, err)
} }
var ingressList IngressList var ingressList IngressList
@ -85,7 +83,7 @@ func (c *clientImpl) GetServices(predicate func(Service) bool) ([]Service, error
body, err := c.do(c.request(getURL)) body, err := c.do(c.request(getURL))
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to create request: GET %q : %v", getURL, err) return nil, fmt.Errorf("failed to create services request: GET %q : %v", getURL, err)
} }
var serviceList ServiceList var serviceList ServiceList
@ -133,22 +131,22 @@ func (c *clientImpl) WatchAll(stopCh <-chan bool) (chan interface{}, chan error,
stopIngresses := make(chan bool) stopIngresses := make(chan bool)
chanIngresses, chanIngressesErr, err := c.WatchIngresses(stopIngresses) chanIngresses, chanIngressesErr, err := c.WatchIngresses(stopIngresses)
if err != nil { if err != nil {
return watchCh, errCh, fmt.Errorf("failed to create watch %v", err) return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
} }
stopServices := make(chan bool) stopServices := make(chan bool)
chanServices, chanServicesErr, err := c.WatchServices(stopServices) chanServices, chanServicesErr, err := c.WatchServices(stopServices)
if err != nil { if err != nil {
return watchCh, errCh, fmt.Errorf("failed to create watch %v", err) return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
} }
stopPods := make(chan bool) stopPods := make(chan bool)
chanPods, chanPodsErr, err := c.WatchPods(stopPods) chanPods, chanPodsErr, err := c.WatchPods(stopPods)
if err != nil { if err != nil {
return watchCh, errCh, fmt.Errorf("failed to create watch %v", err) return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
} }
stopReplicationControllers := make(chan bool) stopReplicationControllers := make(chan bool)
chanReplicationControllers, chanReplicationControllersErr, err := c.WatchReplicationControllers(stopReplicationControllers) chanReplicationControllers, chanReplicationControllersErr, err := c.WatchReplicationControllers(stopReplicationControllers)
if err != nil { if err != nil {
return watchCh, errCh, fmt.Errorf("failed to create watch %v", err) return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
} }
go func() { go func() {
defer close(watchCh) defer close(watchCh)
@ -225,34 +223,35 @@ func (c *clientImpl) watch(url string, stopCh <-chan bool) (chan interface{}, ch
// get version // get version
body, err := c.do(c.request(url)) body, err := c.do(c.request(url))
if err != nil { if err != nil {
return watchCh, errCh, fmt.Errorf("failed to create request: GET %q : %v", url, err) return watchCh, errCh, fmt.Errorf("failed to do version request: GET %q : %v", url, err)
} }
var generic GenericObject var generic GenericObject
if err := json.Unmarshal(body, &generic); err != nil { if err := json.Unmarshal(body, &generic); err != nil {
return watchCh, errCh, fmt.Errorf("failed to create request: GET %q : %v", url, err) return watchCh, errCh, fmt.Errorf("failed to decode version %v", err)
} }
resourceVersion := generic.ResourceVersion resourceVersion := generic.ResourceVersion
url = url + "?watch&resourceVersion=" + resourceVersion url = url + "?watch&resourceVersion=" + resourceVersion
// Make request to Kubernetes API // Make request to Kubernetes API
request := c.request(url) request := c.request(url)
request.Transport.Dial = func(network, addr string) (net.Conn, error) { // request.Transport.Dial = func(network, addr string) (net.Conn, error) {
conn, err := net.Dial(network, addr) // conn, err := net.Dial(network, addr)
if err != nil { // if err != nil {
return nil, err // return nil, err
} // }
// No timeout for long-polling request // // No timeout for long-polling request
conn.SetDeadline(time.Now()) // conn.SetDeadline(time.Now())
return conn, nil // return conn, nil
} // }
req, err := request.TLSClientConfig(c.tls).MakeRequest() req, err := request.MakeRequest()
if err != nil { if err != nil {
return watchCh, errCh, fmt.Errorf("failed to create request: GET %q : %v", url, err) return watchCh, errCh, fmt.Errorf("failed to make watch request: GET %q : %v", url, err)
} }
request.Client.Transport = request.Transport
res, err := request.Client.Do(req) res, err := request.Client.Do(req)
if err != nil { if err != nil {
return watchCh, errCh, fmt.Errorf("failed to make request: GET %q: %v", url, err) return watchCh, errCh, fmt.Errorf("failed to do watch request: GET %q: %v", url, err)
} }
shouldStop := safe.New(false) shouldStop := safe.New(false)