From 2a994e2c766266fed312fdf37319c289fea23bd8 Mon Sep 17 00:00:00 2001 From: andreacappelletti97 Date: Tue, 4 Apr 2023 16:35:40 -0500 Subject: [PATCH] One step closer... --- dax/controller/balancer.go | 6 +++++ dax/controller/balancer/balancer.go | 4 ++-- dax/controller/client/client.go | 2 +- dax/controller/controller.go | 22 ++++++++--------- dax/controller/http/handler.go | 37 ++++++++++++++--------------- dax/controller/sqldb/workerjob.go | 9 ++++++- 6 files changed, 45 insertions(+), 35 deletions(-) diff --git a/dax/controller/balancer.go b/dax/controller/balancer.go index 87edb3cfd..9517b76ec 100644 --- a/dax/controller/balancer.go +++ b/dax/controller/balancer.go @@ -46,6 +46,9 @@ type Balancer interface { // Nodes returns all nodes known by the Balancer. Nodes(tx dax.Transaction) ([]*dax.Node, error) + + // WorkerCount returns the number of workers + WorkerCount(tx dax.Transaction, qdbid dax.QualifiedDatabaseID) (int, error) } // Ensure type implements interface. @@ -94,3 +97,6 @@ func (b *NopBalancer) ReadNode(tx dax.Transaction, addr dax.Address) (*dax.Node, func (b *NopBalancer) Nodes(tx dax.Transaction) ([]*dax.Node, error) { return []*dax.Node{}, nil } +func (b *NopBalancer) WorkerCount(tx dax.Transaction, qdbid dax.QualifiedDatabaseID) (int, error) { + return 0, nil +} diff --git a/dax/controller/balancer/balancer.go b/dax/controller/balancer/balancer.go index 560e6bc29..562ac3f25 100644 --- a/dax/controller/balancer/balancer.go +++ b/dax/controller/balancer/balancer.go @@ -672,8 +672,8 @@ func (b *Balancer) WorkersForTable(tx dax.Transaction, roleType dax.RoleType, qt return workers, nil } -func (b *Balancer) GetDatabaseNumberOfWorkers(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID) (int, error) { - return b.current.WorkerCount(tx, roleType, qdbid) +func (b *Balancer) WorkerCount(tx dax.Transaction, qdbid dax.QualifiedDatabaseID) (int, error) { + return b.current.WorkerCount(tx, "", qdbid) } func (b *Balancer) removeJob(tx dax.Transaction, roleType dax.RoleType, qdbid dax.QualifiedDatabaseID, job dax.Job) (InternalDiffs, error) { diff --git a/dax/controller/client/client.go b/dax/controller/client/client.go index 06b2cf1c0..b848abbd1 100644 --- a/dax/controller/client/client.go +++ b/dax/controller/client/client.go @@ -707,7 +707,7 @@ func (c *Client) GetDatabaseNumberOfWorkers() (int, error) { //url := fmt.Sprintf("%s/database-number-of-workers", "localhost:8080/controller") url := "http://localhost:8080/controller/database-number-of-workers" // print something - fmt.Println("GetDatabaseNumberOfWorkers url: %s", url) + //fmt.Println("GetDatabaseNumberOfWorkers url: %s", url) c.logger.Debugf("GetDatabaseNumberOfWorkers url: %s", url) resp, err := c.httpClient.Get(url) diff --git a/dax/controller/controller.go b/dax/controller/controller.go index 47184abce..1cd0b930a 100644 --- a/dax/controller/controller.go +++ b/dax/controller/controller.go @@ -784,18 +784,16 @@ func (c *Controller) Databases(ctx context.Context, orgID dax.OrganizationID, id return c.Schemar.Databases(tx, orgID, ids...) } -//// get the list of workers being used by a database -//func (c *Controller) GetDatabaseNumberOfWorkers(ctx context.Context, qdbid dax.QualifiedDatabaseID) (int, error) { -// tx, err := c.Transactor.BeginTx(ctx, false) -// if err != nil { -// return -1, errors.Wrap(err, "beginning tx") -// } -// defer tx.Rollback() -// -// // get the number of workers for the database from the balancer -// workers, err := c.Balancer.GetDatabaseNumberOfWorkers(tx, qdbid) -// -//} +// get the list of workers being used by a database +func (c *Controller) GetDatabaseNumberOfWorkers(ctx context.Context, qdbid dax.QualifiedDatabaseID) (int, error) { + tx, err := c.Transactor.BeginTx(ctx, false) + if err != nil { + return 0, errors.Wrap(err, "beginning tx") + } + defer tx.Rollback() + + return c.Balancer.WorkerCount(tx, qdbid) +} // CreateTable adds a table to the schemar, and then sends directives to all // affected nodes based on the change. diff --git a/dax/controller/http/handler.go b/dax/controller/http/handler.go index a37642a6a..ffbbf5549 100644 --- a/dax/controller/http/handler.go +++ b/dax/controller/http/handler.go @@ -226,27 +226,26 @@ func (s *server) patchDatabaseOptions(w http.ResponseWriter, r *http.Request) { // create a response struct func (s *server) getDatabaseNumberOfWorkers(w http.ResponseWriter, r *http.Request) { - //// get the context - //ctx := r.Context() - // - //// create a variable to hold the id - //var id dax.QualifiedDatabaseID - // - //// for now, fill the id with a hard-coded value - //id = dax.QualifiedDatabaseID{ - // OrganizationID: dax.OrganizationID("org1"), - // DatabaseID: dax.DatabaseID("db1"), - //} + // get the context + ctx := r.Context() - // send the id into the controller, get the number of workers back - //numWorkers, err := s.controller.GetDatabaseNumberOfWorkers(ctx, id) - numWorkers := -1 + // create a variable to hold the id + var id dax.QualifiedDatabaseID - //// handle error - //if err != nil { - // http.Error(w, errors.MarshalJSON(err), http.StatusBadRequest) - // return - //} + // for now, fill the id with a hard-coded value + id = dax.QualifiedDatabaseID{ + OrganizationID: dax.OrganizationID("org1"), + DatabaseID: dax.DatabaseID("db1"), + } + + //send the id into the controller, get the number of workers back + numWorkers, err := s.controller.GetDatabaseNumberOfWorkers(ctx, id) + + // handle error + if err != nil { + http.Error(w, errors.MarshalJSON(err), http.StatusBadRequest) + return + } // send the number back to the client if err := json.NewEncoder(w).Encode(numWorkers); err != nil { diff --git a/dax/controller/sqldb/workerjob.go b/dax/controller/sqldb/workerjob.go index 5ca6466f7..e6f826723 100644 --- a/dax/controller/sqldb/workerjob.go +++ b/dax/controller/sqldb/workerjob.go @@ -54,7 +54,14 @@ func (w *workerJobService) WorkerCount(tx dax.Transaction, roleType dax.RoleType return 0, dax.NewErrInvalidTransaction("*sqldb.DaxTransaction") } worker := &models.Worker{} - cnt, err := dt.C.Where("role = ? and database_id = ?", roleType, qdbid.DatabaseID).Count(worker) + query := dt.C.Q() + if roleType != "" { + query = query.Where("role = ?", roleType) + } + if qdbid.DatabaseID != "" { + query = query.Where("database_id = ?", qdbid.DatabaseID) + } + cnt, err := query.Count(worker) return cnt, errors.Wrap(err, "getting count") }