/
info.go
71 lines (61 loc) · 1.74 KB
/
info.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
package storage
import (
"context"
clientv3 "go.etcd.io/etcd/client/v3"
"go.etcd.io/etcd/client/v3/clientv3util"
)
// RegisterBootserver registers a bootserver with etcd database.
func (s Storage) RegisterBootserver(ctx context.Context, lrn int) error {
_, err := s.etcd.Put(ctx, keyBootServer(lrn), "")
return err
}
// DeleteBootServer deletes boot server from etcd database.
func (s Storage) DeleteBootServer(ctx context.Context, lrn int) error {
resp, err := s.etcd.Txn(ctx).
Then(
clientv3.OpDelete(keyBootServer(lrn)),
clientv3.OpDelete(keyInstall(lrn), clientv3.WithPrefix()),
).
Commit()
if err != nil {
return err
}
if resp.Responses[0].GetResponseDeleteRange().Deleted == 0 {
return ErrNotFound
}
return nil
}
// UpdateNecoRelease updates the neco package version with the latest GitHub release.
func (s Storage) UpdateNecoRelease(ctx context.Context, version, leaderKey string) error {
resp, err := s.etcd.Txn(ctx).
If(clientv3util.KeyExists(leaderKey)).
Then(clientv3.OpPut(KeyNecoRelease, version)).
Commit()
if err != nil {
return err
}
if !resp.Succeeded {
return ErrNoLeader
}
return nil
}
// GetNecoRelease gets the neco package version recorded in etcd.
func (s Storage) GetNecoRelease(ctx context.Context) (string, error) {
resp, err := s.etcd.Get(ctx, KeyNecoRelease)
if err != nil {
return "", err
}
if resp.Count == 0 {
return "", nil
}
return string(resp.Kvs[0].Value), nil
}
// WaitInfo waits for update of keys under `info/`
func (s Storage) WaitInfo(ctx context.Context, rev int64) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
ch := s.etcd.Watch(ctx, KeyInfoPrefix,
clientv3.WithPrefix(), clientv3.WithKeysOnly(), clientv3.WithRev(rev+1))
resp := <-ch
return resp.Err()
}