X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/7f4da9388e3d5ec8f38f6d6408916d1d46dfb10f..09cbdc3074b3f1e69c9c537875146f6da0a6ed8f:/lib/dispatchcloud/dispatcher.go diff --git a/lib/dispatchcloud/dispatcher.go b/lib/dispatchcloud/dispatcher.go index 278bcb6657..3403c50c97 100644 --- a/lib/dispatchcloud/dispatcher.go +++ b/lib/dispatchcloud/dispatcher.go @@ -15,13 +15,16 @@ import ( "time" "git.arvados.org/arvados.git/lib/cloud" + "git.arvados.org/arvados.git/lib/controller/dblock" + "git.arvados.org/arvados.git/lib/ctrlctx" "git.arvados.org/arvados.git/lib/dispatchcloud/container" "git.arvados.org/arvados.git/lib/dispatchcloud/scheduler" - "git.arvados.org/arvados.git/lib/dispatchcloud/ssh_executor" + "git.arvados.org/arvados.git/lib/dispatchcloud/sshexecutor" "git.arvados.org/arvados.git/lib/dispatchcloud/worker" "git.arvados.org/arvados.git/sdk/go/arvados" "git.arvados.org/arvados.git/sdk/go/auth" "git.arvados.org/arvados.git/sdk/go/ctxlog" + "git.arvados.org/arvados.git/sdk/go/health" "git.arvados.org/arvados.git/sdk/go/httpserver" "github.com/julienschmidt/httprouter" "github.com/prometheus/client_golang/prometheus" @@ -52,6 +55,7 @@ type dispatcher struct { Registry *prometheus.Registry InstanceSetID cloud.InstanceSetID + dbConnector ctrlctx.DBConnector logger logrus.FieldLogger instanceSet cloud.InstanceSet pool pool @@ -100,7 +104,7 @@ func (disp *dispatcher) Close() { // Make a worker.Executor for the given instance. func (disp *dispatcher) newExecutor(inst cloud.Instance) worker.Executor { - exr := ssh_executor.New(inst) + exr := sshexecutor.New(inst) exr.SetTargetPort(disp.Cluster.Containers.CloudVMs.SSHPort) exr.SetSigners(disp.sshKey) return exr @@ -117,6 +121,7 @@ func (disp *dispatcher) setup() { func (disp *dispatcher) initialize() { disp.logger = ctxlog.FromContext(disp.Context) + disp.dbConnector = ctrlctx.DBConnector{PostgreSQL: disp.Cluster.PostgreSQL} disp.ArvClient.AuthToken = disp.AuthToken @@ -142,6 +147,7 @@ func (disp *dispatcher) initialize() { if err != nil { disp.logger.Fatalf("error initializing driver: %s", err) } + dblock.Dispatch.Lock(disp.Context, disp.dbConnector.GetDB) disp.instanceSet = instanceSet disp.pool = worker.NewPool(disp.logger, disp.ArvClient, disp.Registry, disp.InstanceSetID, disp.instanceSet, disp.newExecutor, disp.sshKey.PublicKey(), disp.Cluster) disp.queue = container.NewQueue(disp.logger, disp.Registry, disp.typeChooser, disp.ArvClient) @@ -164,11 +170,17 @@ func (disp *dispatcher) initialize() { }) mux.Handler("GET", "/metrics", metricsH) mux.Handler("GET", "/metrics.json", metricsH) + mux.Handler("GET", "/_health/:check", &health.Handler{ + Token: disp.Cluster.ManagementToken, + Prefix: "/_health/", + Routes: health.Routes{"ping": disp.CheckHealth}, + }) disp.httpHandler = auth.RequireLiteralToken(disp.Cluster.ManagementToken, mux) } } func (disp *dispatcher) run() { + defer dblock.Dispatch.Unlock() defer close(disp.stopped) defer disp.instanceSet.Stop() defer disp.pool.Stop()