mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-06 00:25:55 +00:00
2843f218bc
* 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
112 lines
2.4 KiB
Go
112 lines
2.4 KiB
Go
package controller
|
|
|
|
import (
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/molecula/featurebase/v3/dax"
|
|
"github.com/molecula/featurebase/v3/errors"
|
|
)
|
|
|
|
// pUnit represents a table/partition combination. As a Stringer, it can be
|
|
// used as a job in the Balancer.
|
|
type pUnit struct {
|
|
t dax.TableKey
|
|
p dax.VersionedPartition
|
|
}
|
|
|
|
func (p pUnit) String() string {
|
|
return fmt.Sprintf("%s|part_%d", p.t, p.p.Num)
|
|
}
|
|
|
|
func (p pUnit) table() dax.TableKey {
|
|
return p.t
|
|
}
|
|
|
|
func (p pUnit) partitionNum() dax.PartitionNum {
|
|
return p.p.Num
|
|
}
|
|
|
|
func partition(t dax.TableKey, p dax.VersionedPartition) pUnit {
|
|
return pUnit{t, p}
|
|
}
|
|
|
|
func partitions(t dax.TableKey, p ...dax.VersionedPartition) []pUnit {
|
|
ret := make([]pUnit, 0, len(p))
|
|
for _, vp := range p {
|
|
ret = append(ret, pUnit{t, vp})
|
|
}
|
|
return ret
|
|
}
|
|
|
|
func decodePartition(j dax.Job) (pUnit, error) {
|
|
s := string(j)
|
|
parts := strings.Split(s, "|")
|
|
if len(parts) != 2 {
|
|
return pUnit{}, errors.Errorf("cannot decode string to partition: %s", s)
|
|
}
|
|
pparts := strings.Split(parts[1], "_")
|
|
if len(pparts) != 2 {
|
|
return pUnit{}, errors.Errorf("cannot decode partition part of string: %s", pparts[1])
|
|
}
|
|
intVar, err := strconv.Atoi(pparts[1])
|
|
if err != nil {
|
|
return pUnit{}, errors.Wrap(err, "converting string to int")
|
|
}
|
|
|
|
return pUnit{
|
|
t: dax.TableKey(parts[0]),
|
|
p: dax.VersionedPartition{
|
|
Num: dax.PartitionNum(intVar),
|
|
Version: -1,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
// sUnit represents a table/shard combination. As a Stringer, it can be used as
|
|
// a job in the Balancer.
|
|
type sUnit struct {
|
|
t dax.TableKey
|
|
s dax.VersionedShard
|
|
}
|
|
|
|
func (s sUnit) String() string {
|
|
return fmt.Sprintf("%s|shard_%s", s.t, s.s.Num)
|
|
}
|
|
|
|
func (s sUnit) table() dax.TableKey {
|
|
return s.t
|
|
}
|
|
|
|
func (s sUnit) shardNum() dax.ShardNum {
|
|
return s.s.Num
|
|
}
|
|
|
|
func shard(t dax.TableKey, s dax.VersionedShard) sUnit {
|
|
return sUnit{t, s}
|
|
}
|
|
|
|
func decodeShard(j dax.Job) (sUnit, error) {
|
|
s := string(j)
|
|
parts := strings.Split(s, "|")
|
|
if len(parts) != 2 {
|
|
return sUnit{}, errors.Errorf("cannot decode string to shardV: %s", s)
|
|
}
|
|
pparts := strings.Split(parts[1], "_")
|
|
if len(pparts) != 2 {
|
|
return sUnit{}, errors.Errorf("cannot decode shard part of string: %s", pparts[1])
|
|
}
|
|
uint64Var, err := strconv.ParseUint(pparts[1], 10, 64)
|
|
if err != nil {
|
|
return sUnit{}, errors.Wrap(err, "converting string to int")
|
|
}
|
|
|
|
return sUnit{
|
|
t: dax.TableKey(parts[0]),
|
|
s: dax.VersionedShard{
|
|
Num: dax.ShardNum(uint64Var),
|
|
Version: -1,
|
|
},
|
|
}, nil
|
|
}
|