featurebase/dax/http/handler.go
Travis Turner 2843f218bc
Introduce ServiceManager and Refactor DAX Integration tests (#2320)
* Introduce ServiceManager and Refactor DAX Integration tests

The ServiceManager provides an interface with which to manage
featurebase (dax) services (mds, queryer, computer). It replaces the
confusing interface implementations in /dax/server/server.go (which
optionally used pointers to in-process objects to satisfy an interface)
with (for now) http implementations. The thought is that even if we're
running all services in-process, we should communicate between services
over http in order to mirror what we would do in a production
environment where the services are running on different nodes.

This batch of commits does quit a lot, most of which is captured here:

- Added `path` support to `dax.Address`. Address is now a string of the form [scheme]://[host]:[port]/[path].
- Added `Holder.directiveApplied` to determine (in tests) if the computer has completed applying the latest directive. This is somewhat temporary until we improve the mds-to-computer logic.
- Removed the "service prefix" code which was prepending client URL paths with the prefix. Instead, the serviceType (mds, queryer, computer[n] is now part of `dax.Address`).
- Removed, from the dax config, the top level `StorageMethod` and `StorageDSN` and now just have `MDS.Config.DataDir`.
- Added `Computer.Config.N` to specify the number of computers to run in-process.
- Moved the `pilosa.MDS` interface to `computer.Registrar`. This is an example of getting the interfaces defined in the right packages.
- Added `SnapshotTable()` method to the mds client (to align with its API).
- Changed `Balancer.AddJob()` to `Balancer.AddJobs()` to support, for example, adding 256 partitions in a single call. Refactored some of the naive Balancer to account for this.
- Added a `Seed` to the top-level config. It's not really useful because of package `crypto/rand`.
- Added an in-memory implementation of the DisCo interface and disabled etcd in a computer service.
- Create sepearte data-dirs for each in-process computer.
- Disabled grpc in dax.
- Modified the sql3 test definition format to support multiple insert steps and separate query results (to align with those steps).

* Changes necessary to get multiple computer instance running in-process

For now the config looks like this:

```
[computer]
run = true
n = 4
```

but we can probably just change that to be something like:

```
[computer]
run = 4
```

*Issues found running multiple "computers" in-process*
- grpc was trying to bind on the same port
  - changed GRPCListener from `*net.TCPListener` to `net.Listener`
  - created a nopListener and set to that for now (i.e. disabled grpc)
- etcd was starting more than once
  - changed dax to use in-memory implementations of the disco interfaces (i.e. stop using etcd)
- IDAllocator (which uses boltdb) was trying to open the `idalloc.db` file more than once
  - realized we have to set separate data-dirs for each holder. that fixed it.

* Port dax integration tests to ManagedCommand

* Modify Balancer-related methods like AddJob to AddJobs

There were (and still are) a lot of places where we were adding on job
at a time, even when we had a long list of jobs to add. This resulted in
every job add (for example adding 1 of 256 shards) taking ~40ms, or over
10s to create a keyed table. One reason was because each job add was
making multiple boltdb transactions.

* Port over more dax integration test stuff

* Add DirectiveApplied to signify that snapshot/writes have loaded.

We use this in tests to avoid using sleeps.
This should be considered temporary; we're going to need a more robust
solution for determining when a computer node is ready to serve complete
data.

* Finish porting dax integration tests

* Improve godocs

* Remove docker-based DAX integration tests.

* go mod tidy

* Move test/managed.go to avoid package conflicts

* Modify IDK integration tests to work with ServiceManager changes

This is really just computer -> computer0
And the MDS DataDir config change.

* cleanup found during review

* echo $CI_COMMIT_REF_SLUG in CI

* remove docker image arg, use build instead
2022-12-05 14:49:17 -06:00

172 lines
3.9 KiB
Go

package http
import (
"context"
"net"
"net/http"
"runtime/debug"
"time"
"github.com/molecula/featurebase/v3/dax/mds"
"github.com/molecula/featurebase/v3/dax/queryer"
"github.com/molecula/featurebase/v3/dax/snapshotter"
"github.com/molecula/featurebase/v3/dax/writelogger"
"github.com/molecula/featurebase/v3/errors"
"github.com/molecula/featurebase/v3/logger"
)
// Handler represents an HTTP handler.
type Handler struct {
Handler http.Handler
bind string
ln net.Listener
// url is used to hold the advertise bind address for printing a log during startup.
url string
closeTimeout time.Duration
server *http.Server
mds *mds.MDS
writeLogger *writelogger.WriteLogger
snapshotter *snapshotter.Snapshotter
queryer *queryer.Queryer
computer http.Handler
logger logger.Logger
}
// HandlerOption is a functional option type for Handler
type HandlerOption func(s *Handler) error
func OptHandlerBind(b string) HandlerOption {
return func(h *Handler) error {
h.bind = b
return nil
}
}
func OptHandlerMDS(m *mds.MDS) HandlerOption {
return func(h *Handler) error {
h.mds = m
return nil
}
}
func OptHandlerWriteLogger(w *writelogger.WriteLogger) HandlerOption {
return func(h *Handler) error {
h.writeLogger = w
return nil
}
}
func OptHandlerSnapshotter(s *snapshotter.Snapshotter) HandlerOption {
return func(h *Handler) error {
h.snapshotter = s
return nil
}
}
func OptHandlerQueryer(q *queryer.Queryer) HandlerOption {
return func(h *Handler) error {
h.queryer = q
return nil
}
}
func OptHandlerLogger(l logger.Logger) HandlerOption {
return func(h *Handler) error {
h.logger = l
return nil
}
}
// OptHandlerCloseTimeout controls how long to wait for the http Server to
// shutdown cleanly before forcibly destroying it. Default is 30 seconds.
func OptHandlerCloseTimeout(d time.Duration) HandlerOption {
return func(h *Handler) error {
h.closeTimeout = d
return nil
}
}
// OptHandlerListener set the listener that will be used by the HTTP server.
// Url must be the advertised URL. It will be used to show a log to the user
// about where the Web UI is. This option is mandatory.
func OptHandlerListener(ln net.Listener, url string) HandlerOption {
return func(h *Handler) error {
h.ln = ln
h.url = url
return nil
}
}
func OptHandlerComputer(handler http.Handler) HandlerOption {
return func(h *Handler) error {
h.computer = handler
return nil
}
}
// NewHandler returns a new instance of Handler with a default logger.
func NewHandler(router http.Handler, opts ...HandlerOption) (*Handler, error) {
handler := &Handler{
logger: logger.NopLogger,
closeTimeout: time.Second * 30,
}
for _, opt := range opts {
err := opt(handler)
if err != nil {
return nil, errors.Wrap(err, "applying option")
}
}
handler.Handler = router
handler.server = &http.Server{Handler: handler}
return handler, nil
}
func (h *Handler) Serve() error {
err := h.server.Serve(h.ln)
if err != nil && err.Error() != "http: Server closed" {
h.logger.Errorf("HTTP handler terminated with error: %s\n", err)
return errors.Wrap(err, "serve http")
}
return nil
}
// Close tries to cleanly shutdown the HTTP server, and failing that, after a
// timeout, calls Server.Close.
func (h *Handler) Close() error {
deadlineCtx, cancelFunc := context.WithDeadline(context.Background(), time.Now().Add(h.closeTimeout))
defer cancelFunc()
err := h.server.Shutdown(deadlineCtx)
if err != nil {
err = h.server.Close()
}
return errors.Wrap(err, "shutdown/close http server")
}
// ServeHTTP handles an HTTP request.
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
defer func() {
if err := recover(); err != nil {
w.WriteHeader(http.StatusInternalServerError)
stack := debug.Stack()
h.logger.Printf("PANIC: %s\n%s", err, stack)
}
}()
h.Handler.ServeHTTP(w, r)
}
// GET /health
func (h *Handler) handleGetHealth(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
}