Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

chore(task): remove task name uniqueness requirement

pull/10616/head
Mark Rushakoff 2018-10-09 09:13:02 -07:00 committed by Mark Rushakoff
parent 0af8de9a3c
commit 35383074f3
3 changed files with 8 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
task/backend/bolt/bolt.go
View File

@ -8,8 +8,6 @@
// bucket(/tasks/v1/org_by_task_id) key(task_id) -> The organization ID (stored as encoded string) associated with given task.
// bucket(/tasks/v1/user_by_task_id) key(:task_id) -> The user ID (stored as encoded string) associated with given task.
// buket(/tasks/v1/name_by_task_id) key(:task_id) -> The user-supplied name of the script.
// bucket(/tasks/v1/name_by_org).bucket(:org_id) key(:task_name) -> Task ID. This allows us to make task names unique for org.
// bucket(/tasks/v1/name_by_user).bucket(:user_id) key(:task_name) -> Task ID. This allows us to make task names unique for user.
// bucket(/tasks/v1/run_ids) -> Counter for run IDs
// bucket(/tasks/v1/orgs).bucket(:org_id) key(:task_id) -> Empty content; presence of :task_id allows for lookup from org to tasks.
// bucket(/tasks/v1/users).bucket(:user_id) key(:task_id) -> Empty content; presence of :task_id allows for lookup from user to tasks.
@ -60,8 +58,6 @@ var (
taskMetaPath = []byte(basePath + "task_meta")
orgByTaskID = []byte(basePath + "org_by_task_id")
userByTaskID = []byte(basePath + "user_by_task_id")
nameByUser = []byte(basePath + "name_by_user")
nameByOrg = []byte(basePath + "name_by_org")
nameByTaskID = []byte(basePath + "name_by_task_id")
runIDs = []byte(basePath + "run_ids")
)
@ -82,7 +78,7 @@ func New(db *bolt.DB, rootBucket string) (*Store, error) {
// create the buckets inside the root
for _, b := range [][]byte{
tasksPath, orgsPath, usersPath, taskMetaPath,
orgByTaskID, userByTaskID, nameByUser, nameByOrg,
orgByTaskID, userByTaskID,
nameByTaskID, runIDs,
} {
_, err := root.CreateBucketIfNotExists(b)
@ -98,30 +94,6 @@ func New(db *bolt.DB, rootBucket string) (*Store, error) {
return &Store{db: db, bucket: bucket}, nil
}
// checkIfNameIsUsed is a helper function that returns an error if a name if already used
func (s *Store) checkIfNameIsUsed(b *bolt.Bucket, name []byte, org, user, taskID platform.ID) error {
var bNameByOrg *bolt.Bucket
if bNameByOrg = b.Bucket(nameByOrg); bNameByOrg != nil {
if bnameByOrgOrg := bNameByOrg.Bucket([]byte(org)); bnameByOrgOrg != nil {
gName := bnameByOrgOrg.Get(name)
if gName != nil && !bytes.Equal(gName, taskID) {
return backend.ErrTaskNameTaken
}
}
}
var bNameByUser *bolt.Bucket
if bNameByUser = b.Bucket(nameByUser); bNameByUser != nil {
if bnameByUserUser := bNameByUser.Bucket([]byte(user)); bnameByUserUser != nil {
gUser := bnameByUserUser.Get(name)
if gUser != nil && !bytes.Equal(gUser, taskID) {
return backend.ErrTaskNameTaken
}
}
}
return nil
}
// CreateTask creates a task in the boltdb task store.
func (s *Store) CreateTask(ctx context.Context, org, user platform.ID, script string, scheduleAfter int64) (platform.ID, error) {
o, err := backend.StoreValidator.CreateArgs(org, user, script)
@ -139,9 +111,6 @@ func (s *Store) CreateTask(ctx context.Context, org, user platform.ID, script st
idi, _ := b.NextSequence() // we ignore this err check, because this can't err inside an Update call
binary.BigEndian.PutUint64(id, idi)
upid = unpadID(id)
if err := s.checkIfNameIsUsed(b, name, org, user, upid); err != nil {
return err
}
// write script
err := b.Bucket(tasksPath).Put(id, []byte(script))
if err != nil {
@ -165,17 +134,6 @@ func (s *Store) CreateTask(ctx context.Context, org, user platform.ID, script st
return err
}
// name by org
orgB, err = b.Bucket(nameByOrg).CreateBucketIfNotExists([]byte(org))
if err != nil {
return err
}
err = orgB.Put(name, upid)
if err != nil {
return err
}
err = b.Bucket(orgByTaskID).Put(id, []byte(org))
if err != nil {
return err
@ -191,16 +149,6 @@ func (s *Store) CreateTask(ctx context.Context, org, user platform.ID, script st
if err != nil {
return err
}
// name by user
userB, err = b.Bucket(nameByUser).CreateBucketIfNotExists([]byte(user))
if err != nil {
return err
}
err = userB.Put(name, upid)
if err != nil {
return err
}
err = b.Bucket(userByTaskID).Put(id, []byte(user))
if err != nil {
@ -246,20 +194,7 @@ func (s *Store) ModifyTask(ctx context.Context, id platform.ID, newScript string
if err != nil {
return err
}
// TODO(docmerlin): org and user should be passed in, somehow, maybe via context or as an arg or something, these lookups are unecessairly expensive
org := b.Bucket(orgByTaskID).Get(paddedID)
user := b.Bucket(userByTaskID).Get(paddedID)
name := []byte(op.Name)
if err := s.checkIfNameIsUsed(b, name, org, user, id); err != nil {
return err
}
if err = b.Bucket(nameByOrg).Bucket(org).Put(name, id); err != nil {
return err
}
if b.Bucket(nameByUser).Bucket(user).Put(name, id); err != nil {
return err
}
return b.Bucket(nameByTaskID).Put(paddedID, name)
return b.Bucket(nameByTaskID).Put(paddedID, []byte(op.Name))
})
}
@ -530,7 +465,6 @@ func (s *Store) DeleteTask(ctx context.Context, id platform.ID) (deleted bool, e
if check := b.Bucket(tasksPath).Get(paddedID); check == nil {
return ErrNotFound
}
name := b.Bucket(nameByTaskID).Get(paddedID)
if err := b.Bucket(taskMetaPath).Delete(paddedID); err != nil {
return err
@ -543,9 +477,6 @@ func (s *Store) DeleteTask(ctx context.Context, id platform.ID) (deleted bool, e
if err := b.Bucket(usersPath).Bucket(user).Delete(paddedID); err != nil {
return err
}
if err := b.Bucket(nameByUser).Bucket(user).Delete(name); err != nil {
return err
}
}
if err := b.Bucket(userByTaskID).Delete(paddedID); err != nil {
return err
@ -559,9 +490,6 @@ func (s *Store) DeleteTask(ctx context.Context, id platform.ID) (deleted bool, e
if err := b.Bucket(orgsPath).Bucket(org).Delete(paddedID); err != nil {
return err
}
if err := b.Bucket(nameByOrg).Bucket(org).Delete(name); err != nil {
return err
}
}
return b.Bucket(orgByTaskID).Delete(paddedID)
})

15
task/backend/inmem_store.go
View File

@ -57,11 +57,6 @@ func (s *inmem) CreateTask(_ context.Context, org, user platform.ID, script stri
s.mu.Lock()
defer s.mu.Unlock()
for _, t := range s.tasks {
if t.Name == task.Name && (bytes.Equal(t.Org, org) || bytes.Equal(t.User, user)) {
return nil, ErrTaskNameTaken
}
}
s.tasks = append(s.tasks, task)
s.runners[id.String()] = StoreTaskMeta{
MaxConcurrency: int32(o.Concurrency),
@ -88,15 +83,7 @@ func (s *inmem) ModifyTask(_ context.Context, id platform.ID, script string) err
continue
}
if t.Name != op.Name {
for i := range s.tasks {
tt := s.tasks[i]
if tt.Name == op.Name && i != n && (bytes.Equal(tt.Org, t.Org) || bytes.Equal(tt.User, t.User)) {
return ErrTaskNameTaken
}
}
t.Name = op.Name
}
t.Name = op.Name
t.Script = script
s.tasks[n] = t
return nil

10
task/backend/storetest/storetest.go
View File

@ -94,9 +94,9 @@ from(bucket:"test") |> range(start:-1h)`
{caseName: "missing user", org: []byte{1}, user: nil, script: script},
{caseName: "missing name", org: []byte{1}, user: []byte{2}, script: scriptNoName},
{caseName: "missing script", org: []byte{1}, user: []byte{2}, script: ""},
{caseName: "repeated name and org", org: []byte{1}, user: []byte{3}, script: script},
{caseName: "repeated name and user", org: []byte{3}, user: []byte{2}, script: script},
{caseName: "repeated name, org, and user", org: []byte{1}, user: []byte{2}, script: script},
{caseName: "repeated name and org", org: []byte{1}, user: []byte{3}, script: script, noerr: true},
{caseName: "repeated name and user", org: []byte{3}, user: []byte{2}, script: script, noerr: true},
{caseName: "repeated name, org, and user", org: []byte{1}, user: []byte{2}, script: script, noerr: true},
} {
t.Run(args.caseName, func(t *testing.T) {
_, err := s.CreateTask(context.Background(), args.org, args.user, args.script, 0)
@ -189,8 +189,8 @@ from(bucket:"y") |> range(start:-1h)`
if err != nil {
t.Fatal(err)
}
if err := s.ModifyTask(context.Background(), id1, script2); err != backend.ErrTaskNameTaken {
t.Fatal("expected ErrTaskNameTaken but did not receive one")
if err := s.ModifyTask(context.Background(), id1, script2); err != nil {
t.Fatalf("expected to be allowed to reuse name when modifying task, but got %v", err)
}
})
}