|
1 | 1 | package metrics |
2 | 2 |
|
3 | 3 | import ( |
| 4 | + "crypto/tls" |
| 5 | + "fmt" |
| 6 | + "net" |
4 | 7 | "net/http" |
5 | 8 | "net/http/pprof" |
| 9 | + "strings" |
6 | 10 |
|
7 | | - "k8s.io/apiserver/pkg/server/healthz" |
8 | | - |
| 11 | + "github.com/cockroachdb/cmux" |
9 | 12 | "github.com/golang/glog" |
10 | 13 | "github.com/prometheus/client_golang/prometheus" |
| 14 | + |
| 15 | + "k8s.io/apiserver/pkg/server/healthz" |
| 16 | + |
| 17 | + "k8s.io/apiserver/pkg/authentication/authenticator" |
| 18 | + "k8s.io/apiserver/pkg/authorization/authorizer" |
11 | 19 | ) |
12 | 20 |
|
| 21 | +type Listener struct { |
| 22 | + Addr string |
| 23 | + |
| 24 | + TLSConfig *tls.Config |
| 25 | + |
| 26 | + Username string |
| 27 | + Password string |
| 28 | + |
| 29 | + Authenticator authenticator.Request |
| 30 | + Authorizer authorizer.Authorizer |
| 31 | + Record authorizer.AttributesRecord |
| 32 | + |
| 33 | + Checks []healthz.HealthzChecker |
| 34 | +} |
| 35 | + |
| 36 | +func (l Listener) handler() http.Handler { |
| 37 | + mux := http.NewServeMux() |
| 38 | + healthz.InstallHandler(mux, l.Checks...) |
| 39 | + |
| 40 | + if l.Authenticator != nil { |
| 41 | + protected := http.NewServeMux() |
| 42 | + protected.HandleFunc("/debug/pprof/", pprof.Index) |
| 43 | + protected.HandleFunc("/debug/pprof/profile", pprof.Profile) |
| 44 | + protected.HandleFunc("/debug/pprof/symbol", pprof.Symbol) |
| 45 | + protected.Handle("/metrics", prometheus.Handler()) |
| 46 | + mux.Handle("/", l.authorizeHandler(protected)) |
| 47 | + } |
| 48 | + return mux |
| 49 | +} |
| 50 | + |
| 51 | +func (l Listener) authorizeHandler(protected http.Handler) http.Handler { |
| 52 | + return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { |
| 53 | + if len(l.Username) > 0 || len(l.Password) > 0 { |
| 54 | + if u, p, ok := req.BasicAuth(); ok { |
| 55 | + if u == l.Username && p == l.Password { |
| 56 | + protected.ServeHTTP(w, req) |
| 57 | + } else { |
| 58 | + http.Error(w, fmt.Sprintf("Unauthorized"), http.StatusUnauthorized) |
| 59 | + } |
| 60 | + return |
| 61 | + } |
| 62 | + } |
| 63 | + |
| 64 | + user, ok, err := l.Authenticator.AuthenticateRequest(req) |
| 65 | + if err != nil { |
| 66 | + glog.V(3).Infof("Unable to authenticate: %v", err) |
| 67 | + http.Error(w, "Unable to authenticate due to an error", http.StatusInternalServerError) |
| 68 | + return |
| 69 | + } |
| 70 | + scopedRecord := l.Record |
| 71 | + switch req.Method { |
| 72 | + case "POST": |
| 73 | + scopedRecord.Verb = "create" |
| 74 | + case "GET", "HEAD": |
| 75 | + scopedRecord.Verb = "get" |
| 76 | + case "PUT": |
| 77 | + scopedRecord.Verb = "update" |
| 78 | + case "PATCH": |
| 79 | + scopedRecord.Verb = "patch" |
| 80 | + case "DELETE": |
| 81 | + scopedRecord.Verb = "delete" |
| 82 | + default: |
| 83 | + scopedRecord.Verb = "" |
| 84 | + } |
| 85 | + switch { |
| 86 | + case req.URL.Path == "/metrics": |
| 87 | + scopedRecord.Subresource = "metrics" |
| 88 | + case strings.HasPrefix(req.URL.Path, "/debug/"): |
| 89 | + scopedRecord.Subresource = "debug" |
| 90 | + } |
| 91 | + scopedRecord.User = user |
| 92 | + ok, reason, err := l.Authorizer.Authorize(scopedRecord) |
| 93 | + if err != nil { |
| 94 | + glog.V(3).Infof("Unable to authenticate: %v", err) |
| 95 | + http.Error(w, "Unable to authenticate due to an error", http.StatusInternalServerError) |
| 96 | + return |
| 97 | + } |
| 98 | + if !ok { |
| 99 | + http.Error(w, fmt.Sprintf("Unauthorized %s", reason), http.StatusUnauthorized) |
| 100 | + return |
| 101 | + } |
| 102 | + protected.ServeHTTP(w, req) |
| 103 | + }) |
| 104 | +} |
| 105 | + |
13 | 106 | // Listen starts a server for health, metrics, and profiling on the provided listen port. |
14 | 107 | // It will terminate the process if the server fails. Metrics and profiling are only exposed |
15 | 108 | // if username and password are provided and the user's input matches. |
16 | | -func Listen(listenAddr string, username, password string, checks ...healthz.HealthzChecker) { |
| 109 | +func (l Listener) Listen() { |
| 110 | + handler := l.handler() |
| 111 | + |
| 112 | + tcpl, err := net.Listen("tcp", l.Addr) |
| 113 | + if err != nil { |
| 114 | + glog.Fatal(err) |
| 115 | + } |
| 116 | + |
| 117 | + // if a TLS connection was requested, set up a connection mux that will send TLS requests to |
| 118 | + // the TLS server but send HTTP requests to the HTTP server. Preserves the ability for HTTP |
| 119 | + // health checks to call HTTP on the router while still allowing TLS certs and end to end |
| 120 | + // metrics protection. |
| 121 | + m := cmux.New(tcpl) |
| 122 | + |
| 123 | + // match HTTP first |
| 124 | + httpl := m.Match(cmux.HTTP1Fast()) |
17 | 125 | go func() { |
18 | | - mux := http.NewServeMux() |
19 | | - healthz.InstallHandler(mux, checks...) |
20 | | - |
21 | | - // TODO: exclude etcd and other unused metrics |
22 | | - |
23 | | - // never enable profiling or metrics without protection |
24 | | - if len(username) > 0 && len(password) > 0 { |
25 | | - protected := http.NewServeMux() |
26 | | - protected.HandleFunc("/debug/pprof/", pprof.Index) |
27 | | - protected.HandleFunc("/debug/pprof/profile", pprof.Profile) |
28 | | - protected.HandleFunc("/debug/pprof/symbol", pprof.Symbol) |
29 | | - protected.Handle("/metrics", prometheus.Handler()) |
30 | | - mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) { |
31 | | - if u, p, ok := req.BasicAuth(); !ok || u != username || p != password { |
32 | | - http.Error(w, "Unauthorized", http.StatusUnauthorized) |
33 | | - return |
34 | | - } |
35 | | - protected.ServeHTTP(w, req) |
36 | | - }) |
| 126 | + s := &http.Server{ |
| 127 | + Handler: handler, |
37 | 128 | } |
38 | | - |
39 | | - server := &http.Server{ |
40 | | - Addr: listenAddr, |
41 | | - Handler: mux, |
| 129 | + if err := s.Serve(httpl); err != cmux.ErrListenerClosed { |
| 130 | + glog.Fatal(err) |
42 | 131 | } |
43 | | - glog.Infof("Router health and metrics port listening at %s", listenAddr) |
44 | | - glog.Fatal(server.ListenAndServe()) |
45 | 132 | }() |
46 | 133 |
|
| 134 | + // match TLS if configured |
| 135 | + if l.TLSConfig != nil { |
| 136 | + glog.Infof("Router health and metrics port listening at %s on HTTP and HTTPS", l.Addr) |
| 137 | + tlsl := m.Match(cmux.Any()) |
| 138 | + tlsl = tls.NewListener(tlsl, l.TLSConfig) |
| 139 | + go func() { |
| 140 | + s := &http.Server{ |
| 141 | + Handler: handler, |
| 142 | + } |
| 143 | + if err := s.Serve(tlsl); err != cmux.ErrListenerClosed { |
| 144 | + glog.Fatal(err) |
| 145 | + } |
| 146 | + }() |
| 147 | + } else { |
| 148 | + glog.Infof("Router health and metrics port listening at %s", l.Addr) |
| 149 | + } |
| 150 | + |
| 151 | + go func() { |
| 152 | + if err := m.Serve(); !strings.Contains(err.Error(), "use of closed network connection") { |
| 153 | + glog.Fatal(err) |
| 154 | + } |
| 155 | + }() |
47 | 156 | } |
0 commit comments