scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

One step closer...

Browse source
This commit is contained in:
andreacappelletti97 2023-04-04 16:35:40 -05:00
parent 8e15351b2b
commit 2a994e2c76
6 changed files with 45 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
dax/controller/balancer.go
View file

@ -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
}

4
dax/controller/balancer/balancer.go
View file

@ -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) {

2
dax/controller/client/client.go
View file

@ -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)

22
dax/controller/controller.go
View file

@ -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.

37
dax/controller/http/handler.go
View file

@ -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 {

9
dax/controller/sqldb/workerjob.go
View file

@ -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")
}