19166: Merge branch 'main'
[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.Inspect(reg, cluster.ManagementToken,
151                                         httpserver.LogRequests(
152                                                 interceptHealthReqs(cluster.ManagementToken, handler.CheckHealth,
153                                                         httpserver.NewRequestLimiter(cluster.API.MaxConcurrentRequests, handler, reg)))))))
154         srv := &httpserver.Server{
155                 Server: http.Server{
156                         Handler:     ifCollectionInHost(instrumented, instrumented.ServeAPI(cluster.ManagementToken, instrumented)),
157                         BaseContext: func(net.Listener) context.Context { return ctx },
158                 },
159                 Addr: listenURL.Host,
160         }
161         if listenURL.Scheme == "https" || listenURL.Scheme == "wss" {
162                 tlsconfig, err := makeTLSConfig(cluster, logger)
163                 if err != nil {
164                         logger.WithError(err).Errorf("cannot start %s service on %s", c.svcName, listenURL.String())
165                         return 1
166                 }
167                 srv.TLSConfig = tlsconfig
168         }
169         err = srv.Start()
170         if err != nil {
171                 return 1
172         }
173         logger.WithFields(logrus.Fields{
174                 "URL":     listenURL,
175                 "Listen":  srv.Addr,
176                 "Service": c.svcName,
177                 "Version": cmd.Version.String(),
178         }).Info("listening")
179         if _, err := daemon.SdNotify(false, "READY=1"); err != nil {
180                 logger.WithError(err).Errorf("error notifying init daemon")
181         }
182         go func() {
183                 // Shut down server if caller cancels context
184                 <-ctx.Done()
185                 srv.Close()
186         }()
187         go func() {
188                 // Shut down server if handler dies
189                 <-handler.Done()
190                 srv.Close()
191         }()
192         err = srv.Wait()
193         if err != nil {
194                 return 1
195         }
196         return 0
197 }
198
199 // If an incoming request's target vhost has an embedded collection
200 // UUID or PDH, handle it with hTrue, otherwise handle it with
201 // hFalse.
202 //
203 // Facilitates routing "http://collections.example/metrics" to metrics
204 // and "http://{uuid}.collections.example/metrics" to a file in a
205 // collection.
206 func ifCollectionInHost(hTrue, hFalse http.Handler) http.Handler {
207         return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
208                 if arvados.CollectionIDFromDNSName(r.Host) != "" {
209                         hTrue.ServeHTTP(w, r)
210                 } else {
211                         hFalse.ServeHTTP(w, r)
212                 }
213         })
214 }
215
216 func interceptHealthReqs(mgtToken string, checkHealth func() error, next http.Handler) http.Handler {
217         mux := httprouter.New()
218         mux.Handler("GET", "/_health/ping", &health.Handler{
219                 Token:  mgtToken,
220                 Prefix: "/_health/",
221                 Routes: health.Routes{"ping": checkHealth},
222         })
223         mux.NotFound = next
224         return ifCollectionInHost(next, mux)
225 }
226
227 // Determine listenURL (addr:port where server should bind) and
228 // internalURL (target url that client should connect to) for a
229 // service.
230 //
231 // If the config does not specify ListenURL, we check all of the
232 // configured InternalURLs. If there is exactly one that matches our
233 // hostname, or exactly one that matches a local interface address,
234 // then we use that as listenURL.
235 //
236 // Note that listenURL and internalURL may use different protocols
237 // (e.g., listenURL is http, but the service sits behind a proxy, so
238 // clients connect using https).
239 func getListenAddr(svcs arvados.Services, prog arvados.ServiceName, log logrus.FieldLogger) (arvados.URL, arvados.URL, error) {
240         svc, ok := svcs.Map()[prog]
241         if !ok {
242                 return arvados.URL{}, arvados.URL{}, fmt.Errorf("unknown service name %q", prog)
243         }
244
245         if want := os.Getenv("ARVADOS_SERVICE_INTERNAL_URL"); want != "" {
246                 url, err := url.Parse(want)
247                 if err != nil {
248                         return arvados.URL{}, arvados.URL{}, fmt.Errorf("$ARVADOS_SERVICE_INTERNAL_URL (%q): %s", want, err)
249                 }
250                 if url.Path == "" {
251                         url.Path = "/"
252                 }
253                 for internalURL, conf := range svc.InternalURLs {
254                         if internalURL.String() == url.String() {
255                                 listenURL := conf.ListenURL
256                                 if listenURL.Host == "" {
257                                         listenURL = internalURL
258                                 }
259                                 return listenURL, internalURL, nil
260                         }
261                 }
262                 log.Warnf("possible configuration error: listening on %s (from $ARVADOS_SERVICE_INTERNAL_URL) even though configuration does not have a matching InternalURLs entry", url)
263                 internalURL := arvados.URL(*url)
264                 return internalURL, internalURL, nil
265         }
266
267         errors := []string{}
268         for internalURL, conf := range svc.InternalURLs {
269                 listenURL := conf.ListenURL
270                 if listenURL.Host == "" {
271                         // If ListenURL is not specified, assume
272                         // InternalURL is also usable as the listening
273                         // proto/addr/port (i.e., simple case with no
274                         // intermediate proxy/routing)
275                         listenURL = internalURL
276                 }
277                 listenAddr := listenURL.Host
278                 if _, _, err := net.SplitHostPort(listenAddr); err != nil {
279                         // url "https://foo.example/" (with no
280                         // explicit port name/number) means listen on
281                         // the well-known port for the specified
282                         // protocol, "foo.example:https".
283                         port := listenURL.Scheme
284                         if port == "ws" || port == "wss" {
285                                 port = "http" + port[2:]
286                         }
287                         listenAddr = net.JoinHostPort(listenAddr, port)
288                 }
289                 listener, err := net.Listen("tcp", listenAddr)
290                 if err == nil {
291                         listener.Close()
292                         return listenURL, internalURL, nil
293                 } else if strings.Contains(err.Error(), "cannot assign requested address") {
294                         // If 'Host' specifies a different server than
295                         // the current one, it'll resolve the hostname
296                         // to IP address, and then fail because it
297                         // can't bind an IP address it doesn't own.
298                         continue
299                 } else {
300                         errors = append(errors, fmt.Sprintf("%s: %s", listenURL, err))
301                 }
302         }
303         if len(errors) > 0 {
304                 return arvados.URL{}, arvados.URL{}, fmt.Errorf("could not enable the %q service on this host: %s", prog, strings.Join(errors, "; "))
305         }
306         return arvados.URL{}, arvados.URL{}, fmt.Errorf("configuration does not enable the %q service on this host", prog)
307 }
308
309 type contextKeyURL struct{}
310
311 func URLFromContext(ctx context.Context) (arvados.URL, bool) {
312         u, ok := ctx.Value(contextKeyURL{}).(arvados.URL)
313         return u, ok
314 }