Skip to content

Commit b3fbb2b

Browse files
author
Aaron Schlesinger
committed
fix(pkg/healthsrv): do the 3 health checks concurrently
1 parent 4808fe1 commit b3fbb2b

4 files changed

Lines changed: 131 additions & 36 deletions

File tree

‎pkg/healthsrv/buckets_lister.go‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,3 +24,21 @@ type errBucketLister struct {
2424
func (e errBucketLister) ListBuckets(*s3.ListBucketsInput) (*s3.ListBucketsOutput, error) {
2525
return nil, e.err
2626
}
27+
28+
// listBuckets calls bl.ListBuckets(...) and sends the results back on the various given channels. This func is intended to be run in a goroutine and communicates via the channels it's passed.
29+
//
30+
// On success, it passes the bucket output on succCh, and on failure, it passes the error on errCh. At most one of {succCh, errCh} will be sent on. If stopCh is closed, no pending or future sends will occur.
31+
func listBuckets(bl BucketLister, succCh chan<- *s3.ListBucketsOutput, errCh chan<- error, stopCh <-chan struct{}) {
32+
lbOut, err := bl.ListBuckets(&s3.ListBucketsInput{})
33+
if err != nil {
34+
select {
35+
case errCh <- err:
36+
case <-stopCh:
37+
}
38+
return
39+
}
40+
select {
41+
case succCh <- lbOut:
42+
case <-stopCh:
43+
}
44+
}

‎pkg/healthsrv/circuit_state.go‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
package healthsrv
2+
3+
import (
4+
"fmt"
5+
6+
"github.com/deis/builder/pkg/sshd"
7+
)
8+
9+
// circuitState determines whether circ.State() == sshd.ClosedState, and sends the results back on the various given channels. This func is intended to be run in a goroutine and communicates via the channels it's passed.
10+
//
11+
// If the circuit is closed, it passes an empty struct back on succCh. On failure, it sends an error back on errCh. At most one of {succCh, errCh} will be sent on. If stopCh is closed, no pending or future sends will occur.
12+
func circuitState(circ *sshd.Circuit, succCh chan<- struct{}, errCh chan<- error, stopCh <-chan struct{}) {
13+
// There's a race between the boolean eval and the HTTP error returned (the circuit could close between the two). This function should be polled to avoid that problem. If it's being used in a k8s probe, then you're fine because k8s will repeat the health probe and effectively re-evaluate the boolean
14+
if circ.State() != sshd.ClosedState {
15+
select {
16+
case errCh <- fmt.Errorf("SSH Server is not yet started"):
17+
case <-stopCh:
18+
}
19+
return
20+
}
21+
select {
22+
case succCh <- struct{}{}:
23+
case <-stopCh:
24+
}
25+
}

‎pkg/healthsrv/healthz_handler.go‎

Lines changed: 69 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,7 @@ import (
99
s3 "github.com/aws/aws-sdk-go/service/s3"
1010
"github.com/deis/builder/pkg/gitreceive/log"
1111
"github.com/deis/builder/pkg/sshd"
12-
"k8s.io/kubernetes/pkg/fields"
13-
"k8s.io/kubernetes/pkg/labels"
12+
"k8s.io/kubernetes/pkg/api"
1413
)
1514

1615
type healthZRespBucket struct {
@@ -31,43 +30,77 @@ type healthZResp struct {
3130
SSHServerStarted bool `json:"ssh_server_started"`
3231
}
3332

33+
func marshalHealthZResp(w http.ResponseWriter, rsp healthZResp) {
34+
if err := json.NewEncoder(w).Encode(rsp); err != nil {
35+
str := fmt.Sprintf("Error encoding JSON (%s)", err)
36+
http.Error(w, str, http.StatusInternalServerError)
37+
return
38+
}
39+
}
40+
3441
func healthZHandler(nsLister NamespaceLister, bLister BucketLister, serverCircuit *sshd.Circuit) http.Handler {
3542
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
36-
// There's a race between the boolean eval and the HTTP error returned (the server could start up between the two), but k8s will repeat the health probe request and effectively re-evaluate the boolean. The result is that the server may not start until the next probe in those cases
37-
if serverCircuit.State() != sshd.ClosedState {
38-
str := fmt.Sprintf("SSH Server is not yet started")
39-
log.Err(str)
40-
http.Error(w, str, http.StatusServiceUnavailable)
41-
return
42-
}
43-
lbOut, err := bLister.ListBuckets(&s3.ListBucketsInput{})
44-
if err != nil {
45-
str := fmt.Sprintf("Error listing buckets (%s)", err)
46-
log.Err(str)
47-
http.Error(w, str, http.StatusServiceUnavailable)
48-
return
49-
}
50-
var rsp healthZResp
51-
for _, buck := range lbOut.Buckets {
52-
rsp.S3Buckets = append(rsp.S3Buckets, convertBucket(buck))
53-
}
43+
stopCh := make(chan struct{})
5444

55-
nsList, err := nsLister.List(labels.Everything(), fields.Everything())
56-
if err != nil {
57-
str := fmt.Sprintf("Error listing namespaces (%s)", err)
58-
log.Err(str)
59-
http.Error(w, str, http.StatusServiceUnavailable)
60-
return
61-
}
62-
for _, ns := range nsList.Items {
63-
rsp.Namespaces = append(rsp.Namespaces, ns.Name)
64-
}
65-
rsp.SSHServerStarted = true
66-
if err := json.NewEncoder(w).Encode(rsp); err != nil {
67-
str := fmt.Sprintf("Error encoding JSON (%s)", err)
68-
http.Error(w, str, http.StatusInternalServerError)
69-
return
45+
serverStateCh := make(chan struct{})
46+
serverStateErrCh := make(chan error)
47+
go circuitState(serverCircuit, serverStateCh, serverStateErrCh, stopCh)
48+
49+
listBucketsCh := make(chan *s3.ListBucketsOutput)
50+
listBucketsErrCh := make(chan error)
51+
go listBuckets(bLister, listBucketsCh, listBucketsErrCh, stopCh)
52+
53+
namespaceListerCh := make(chan *api.NamespaceList)
54+
namespaceListerErrCh := make(chan error)
55+
go listNamespaces(nsLister, namespaceListerCh, namespaceListerErrCh, stopCh)
56+
57+
var rsp healthZResp
58+
serverState, bucketState, namespaceState := false, false, false
59+
for {
60+
select {
61+
case err := <-serverStateErrCh:
62+
log.Err(err.Error())
63+
http.Error(w, err.Error(), http.StatusServiceUnavailable)
64+
close(stopCh)
65+
return
66+
case err := <-listBucketsErrCh:
67+
str := fmt.Sprintf("Error listing buckets (%s)", err)
68+
log.Err(str)
69+
http.Error(w, str, http.StatusServiceUnavailable)
70+
close(stopCh)
71+
return
72+
case err := <-namespaceListerErrCh:
73+
str := fmt.Sprintf("Error listing namespaces (%s)", err)
74+
log.Err(str)
75+
http.Error(w, str, http.StatusServiceUnavailable)
76+
close(stopCh)
77+
return
78+
case <-serverStateCh:
79+
serverState = true
80+
rsp.SSHServerStarted = true
81+
if serverState && bucketState && namespaceState {
82+
marshalHealthZResp(w, rsp)
83+
return
84+
}
85+
case lbOut := <-listBucketsCh:
86+
bucketState = true
87+
for _, buck := range lbOut.Buckets {
88+
rsp.S3Buckets = append(rsp.S3Buckets, convertBucket(buck))
89+
}
90+
if serverState && bucketState && namespaceState {
91+
marshalHealthZResp(w, rsp)
92+
return
93+
}
94+
case nsList := <-namespaceListerCh:
95+
namespaceState = true
96+
for _, ns := range nsList.Items {
97+
rsp.Namespaces = append(rsp.Namespaces, ns.Name)
98+
}
99+
if serverState && bucketState && namespaceState {
100+
marshalHealthZResp(w, rsp)
101+
return
102+
}
103+
}
70104
}
71-
// TODO: check server is running
72105
})
73106
}

‎pkg/healthsrv/namespace_lister.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,3 +25,22 @@ type errNamespaceLister struct {
2525
func (e errNamespaceLister) List(labels.Selector, fields.Selector) (*api.NamespaceList, error) {
2626
return nil, e.err
2727
}
28+
29+
// listNamespaces calls nl.List(...) and sends the results back on the various given channels. This func is intended to be run in a goroutine and communicates via the channels it's passed.
30+
//
31+
// On success, it passes the namespace list on succCh, and on failure, it passes the error on errCh. At most one of {succCh, errCh} will be sent on. If stopCh is closed, no pending or future sends will occur.
32+
func listNamespaces(nl NamespaceLister, succCh chan<- *api.NamespaceList, errCh chan<- error, stopCh <-chan struct{}) {
33+
nsList, err := nl.List(labels.Everything(), fields.Everything())
34+
if err != nil {
35+
select {
36+
case errCh <- err:
37+
case <-stopCh:
38+
}
39+
return
40+
}
41+
select {
42+
case succCh <- nsList:
43+
case <-stopCh:
44+
return
45+
}
46+
}

0 commit comments

Comments
 (0)