16561: Detect ambiguous uses of ListenAddress.
[arvados.git] / lib / service / cmd.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: Apache-2.0
4
5 // Package service provides a cmd.Handler that brings up a system service.
6 package service
7
8 import (
9         "context"
10         "flag"
11         "fmt"
12         "io"
13         "net"
14         "net/http"
15         _ "net/http/pprof"
16         "net/url"
17         "os"
18         "strings"
19
20         "git.arvados.org/arvados.git/lib/cmd"
21         "git.arvados.org/arvados.git/lib/config"
22         "git.arvados.org/arvados.git/sdk/go/arvados"
23         "git.arvados.org/arvados.git/sdk/go/ctxlog"
24         "git.arvados.org/arvados.git/sdk/go/health"
25         "git.arvados.org/arvados.git/sdk/go/httpserver"
26         "github.com/coreos/go-systemd/daemon"
27         "github.com/julienschmidt/httprouter"
28         "github.com/prometheus/client_golang/prometheus"
29         "github.com/sirupsen/logrus"
30 )
31
32 type Handler interface {
33         http.Handler
34         CheckHealth() error
35         // Done returns a channel that closes when the handler shuts
36         // itself down, or nil if this never happens.
37         Done() <-chan struct{}
38 }
39
40 type NewHandlerFunc func(_ context.Context, _ *arvados.Cluster, token string, registry *prometheus.Registry) Handler
41
42 type command struct {
43         newHandler NewHandlerFunc
44         svcName    arvados.ServiceName
45         ctx        context.Context // enables tests to shutdown service; no public API yet
46 }
47
48 // Command returns a cmd.Handler that loads site config, calls
49 // newHandler with the current cluster and node configs, and brings up
50 // an http server with the returned handler.
51 //
52 // The handler is wrapped with server middleware (adding X-Request-ID
53 // headers, logging requests/responses, etc).
54 func Command(svcName arvados.ServiceName, newHandler NewHandlerFunc) cmd.Handler {
55         return &command{
56                 newHandler: newHandler,
57                 svcName:    svcName,
58                 ctx:        context.Background(),
59         }
60 }
61
62 func (c *command) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
63         log := ctxlog.New(stderr, "json", "info")
64
65         var err error
66         defer func() {
67                 if err != nil {
68                         log.WithError(err).Error("exiting")
69                 }
70         }()
71
72         flags := flag.NewFlagSet("", flag.ContinueOnError)
73         flags.SetOutput(stderr)
74
75         loader := config.NewLoader(stdin, log)
76         loader.SetupFlags(flags)
77
78         // prog is [keepstore, keep-web, git-httpd, ...]  but the
79         // legacy config flags are [-legacy-keepstore-config,
80         // -legacy-keepweb-config, -legacy-git-httpd-config, ...]
81         legacyFlag := "-legacy-" + strings.Replace(prog, "keep-", "keep", 1) + "-config"
82         args = loader.MungeLegacyConfigArgs(log, args, legacyFlag)
83
84         versionFlag := flags.Bool("version", false, "Write version information to stdout and exit 0")
85         pprofAddr := flags.String("pprof", "", "Serve Go profile data at `[addr]:port`")
86         if ok, code := cmd.ParseFlags(flags, prog, args, "", stderr); !ok {
87                 return code
88         } else if *versionFlag {
89                 return cmd.Version.RunCommand(prog, args, stdin, stdout, stderr)
90         }
91
92         if *pprofAddr != "" {
93                 go func() {
94                         log.Println(http.ListenAndServe(*pprofAddr, nil))
95                 }()
96         }
97
98         if strings.HasSuffix(prog, "controller") {
99                 // Some config-loader checks try to make API calls via
100                 // controller. Those can't be expected to work if this
101                 // process _is_ the controller: we haven't started an
102                 // http server yet.
103                 loader.SkipAPICalls = true
104         }
105
106         cfg, err := loader.Load()
107         if err != nil {
108                 return 1
109         }
110         cluster, err := cfg.GetCluster("")
111         if err != nil {
112                 return 1
113         }
114
115         // Now that we've read the config, replace the bootstrap
116         // logger with a new one according to the logging config.
117         log = ctxlog.New(stderr, cluster.SystemLogs.Format, cluster.SystemLogs.LogLevel)
118         logger := log.WithFields(logrus.Fields{
119                 "PID":       os.Getpid(),
120                 "ClusterID": cluster.ClusterID,
121         })
122         ctx := ctxlog.Context(c.ctx, logger)
123
124         listenURL, internalURL, err := getListenAddr(cluster.Services, c.svcName, log)
125         if err != nil {
126                 return 1
127         }
128         ctx = context.WithValue(ctx, contextKeyURL{}, internalURL)
129
130         reg := prometheus.NewRegistry()
131         loader.RegisterMetrics(reg)
132
133         // arvados_version_running{version="1.2.3~4"} 1.0
134         mVersion := prometheus.NewGaugeVec(prometheus.GaugeOpts{
135                 Namespace: "arvados",
136                 Name:      "version_running",
137                 Help:      "Indicated version is running.",
138         }, []string{"version"})
139         mVersion.WithLabelValues(cmd.Version.String()).Set(1)
140         reg.MustRegister(mVersion)
141
142         handler := c.newHandler(ctx, cluster, cluster.SystemRootToken, reg)
143         if err = handler.CheckHealth(); err != nil {
144                 return 1
145         }
146
147         instrumented := httpserver.Instrument(reg, log,
148                 httpserver.HandlerWithDeadline(cluster.API.RequestTimeout.Duration(),
149                         httpserver.AddRequestIDs(
150                                 httpserver.LogRequests(
151                                         interceptHealthReqs(cluster.ManagementToken, handler.CheckHealth,
152                                                 httpserver.NewRequestLimiter(cluster.API.MaxConcurrentRequests, handler, reg))))))
153         srv := &httpserver.Server{
154                 Server: http.Server{
155                         Handler:     ifCollectionInHost(instrumented, instrumented.ServeAPI(cluster.ManagementToken, instrumented)),
156                         BaseContext: func(net.Listener) context.Context { return ctx },
157                 },
158                 Addr: listenURL.Host,
159         }
160         if listenURL.Scheme == "https" || listenURL.Scheme == "wss" {
161                 tlsconfig, err := tlsConfigWithCertUpdater(cluster, logger)
162                 if err != nil {
163                         logger.WithError(err).Errorf("cannot start %s service on %s", c.svcName, listenURL.String())
164                         return 1
165                 }
166                 srv.TLSConfig = tlsconfig
167         }
168         err = srv.Start()
169         if err != nil {
170                 return 1
171         }
172         logger.WithFields(logrus.Fields{
173                 "URL":     listenURL,
174                 "Listen":  srv.Addr,
175                 "Service": c.svcName,
176                 "Version": cmd.Version.String(),
177         }).Info("listening")
178         if _, err := daemon.SdNotify(false, "READY=1"); err != nil {
179                 logger.WithError(err).Errorf("error notifying init daemon")
180         }
181         go func() {
182                 // Shut down server if caller cancels context
183                 <-ctx.Done()
184                 srv.Close()
185         }()
186         go func() {
187                 // Shut down server if handler dies
188                 <-handler.Done()
189                 srv.Close()
190         }()
191         err = srv.Wait()
192         if err != nil {
193                 return 1
194         }
195         return 0
196 }
197
198 // If an incoming request's target vhost has an embedded collection
199 // UUID or PDH, handle it with hTrue, otherwise handle it with
200 // hFalse.
201 //
202 // Facilitates routing "http://collections.example/metrics" to metrics
203 // and "http://{uuid}.collections.example/metrics" to a file in a
204 // collection.
205 func ifCollectionInHost(hTrue, hFalse http.Handler) http.Handler {
206         return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
207                 if arvados.CollectionIDFromDNSName(r.Host) != "" {
208                         hTrue.ServeHTTP(w, r)
209                 } else {
210                         hFalse.ServeHTTP(w, r)
211                 }
212         })
213 }
214
215 func interceptHealthReqs(mgtToken string, checkHealth func() error, next http.Handler) http.Handler {
216         mux := httprouter.New()
217         mux.Handler("GET", "/_health/ping", &health.Handler{
218                 Token:  mgtToken,
219                 Prefix: "/_health/",
220                 Routes: health.Routes{"ping": checkHealth},
221         })
222         mux.NotFound = next
223         return ifCollectionInHost(next, mux)
224 }
225
226 func getListenAddr(svcs arvados.Services, prog arvados.ServiceName, log logrus.FieldLogger) (arvados.URL, arvados.URL, error) {
227         svc, ok := svcs.Map()[prog]
228         if !ok {
229                 return arvados.URL{}, arvados.URL{}, fmt.Errorf("unknown service name %q", prog)
230         }
231
232         if want := os.Getenv("ARVADOS_SERVICE_INTERNAL_URL"); want == "" {
233         } else if url, err := url.Parse(want); err != nil {
234                 return arvados.URL{}, arvados.URL{}, fmt.Errorf("$ARVADOS_SERVICE_INTERNAL_URL (%q): %s", want, err)
235         } else {
236                 if url.Path == "" {
237                         url.Path = "/"
238                 }
239                 internalURL := arvados.URL(*url)
240                 listenURL := arvados.URL(*url)
241                 if svc.ListenAddress != "" {
242                         listenURL.Host = svc.ListenAddress
243                 }
244                 return listenURL, internalURL, nil
245         }
246
247         if svc.ListenAddress != "" {
248                 scheme := ""
249                 for internalURL := range svc.InternalURLs {
250                         if internalURL.Host == svc.ListenAddress {
251                                 if len(svc.InternalURLs) > 1 {
252                                         log.Warnf("possible configuration error: multiple InternalURLs entries exist for %s but only %q will ever be used because it matches ListenAddress", prog, internalURL.String())
253                                 }
254                                 return internalURL, internalURL, nil
255                         }
256                         switch scheme {
257                         case "":
258                                 scheme = internalURL.Scheme
259                         case internalURL.Scheme:
260                         default:
261                                 scheme = "-" // different InternalURLs have different schemes
262                         }
263                 }
264                 if scheme == "-" {
265                         return arvados.URL{}, arvados.URL{}, fmt.Errorf("cannot use ListenAddress %q: InternalURLs use multiple schemes and none have host %q", svc.ListenAddress, svc.ListenAddress)
266                 }
267                 if scheme == "" {
268                         // No entries at all in InternalURLs
269                         scheme = "http"
270                 }
271                 listenURL := arvados.URL{}
272                 listenURL.Host = svc.ListenAddress
273                 listenURL.Scheme = scheme
274                 listenURL.Path = "/"
275                 return listenURL, listenURL, nil
276         }
277
278         errors := []string{}
279         for internalURL := range svc.InternalURLs {
280                 listener, err := net.Listen("tcp", internalURL.Host)
281                 if err == nil {
282                         listener.Close()
283                         return internalURL, internalURL, nil
284                 } else if strings.Contains(err.Error(), "cannot assign requested address") {
285                         // If 'Host' specifies a different server than
286                         // the current one, it'll resolve the hostname
287                         // to IP address, and then fail because it
288                         // can't bind an IP address it doesn't own.
289                         continue
290                 } else {
291                         errors = append(errors, fmt.Sprintf("tried %v, got %v", internalURL, err))
292                 }
293         }
294         if len(errors) > 0 {
295                 return arvados.URL{}, arvados.URL{}, fmt.Errorf("could not enable the %q service on this host: %s", prog, strings.Join(errors, "; "))
296         }
297         return arvados.URL{}, arvados.URL{}, fmt.Errorf("configuration does not enable the %q service on this host", prog)
298 }
299
300 type contextKeyURL struct{}
301
302 func URLFromContext(ctx context.Context) (arvados.URL, bool) {
303         u, ok := ctx.Value(contextKeyURL{}).(arvados.URL)
304         return u, ok
305 }