From 93fb332b84bbe07735034e835035583bd16602a5 Mon Sep 17 00:00:00 2001 From: Victor Vrantchan Date: Tue, 11 Sep 2018 02:01:25 +0000 Subject: [PATCH] Depsync cleanup (#505) Closes #302 This package still needs a lot of work, but I feel like this is a good stopping point. In a follow up, I plan on writing some tests, and refactoring some of the ways the sync.Watcher goroutines are managed. --- cmd/mdmctl/apply_dep_autoassigner.go | 4 +- cmd/mdmctl/apply_dep_profile.go | 4 +- cmd/mdmctl/setup.go | 6 +- cmd/micromdm/serve.go | 6 +- dep/depsync/config.go | 95 ---------- dep/depsync/service.go | 16 -- go.sum | 1 + .../dep/sync}/apply_autoassigner.go | 7 +- platform/dep/sync/builtin/db.go | 116 ++++++++++++ {dep/depsync => platform/dep/sync}/client.go | 2 +- {dep/depsync => platform/dep/sync}/depsync.go | 179 ++++++++---------- {dep/depsync => platform/dep/sync}/event.go | 4 +- .../dep/sync}/get_autoassigners.go | 15 +- .../sync}/internal/depsyncproto/depsync.go | 0 .../sync}/internal/depsyncproto/depsync.pb.go | 0 .../sync}/internal/depsyncproto/depsync.proto | 0 .../dep/sync}/remove_autoassigners.go | 7 +- {dep/depsync => platform/dep/sync}/server.go | 6 +- platform/dep/sync/service.go | 40 ++++ {dep/depsync => platform/dep/sync}/syncnow.go | 2 +- platform/device/worker.go | 10 +- server/server.go | 22 ++- 22 files changed, 289 insertions(+), 253 deletions(-) delete mode 100644 dep/depsync/config.go delete mode 100644 dep/depsync/service.go rename {dep/depsync => platform/dep/sync}/apply_autoassigner.go (92%) create mode 100644 platform/dep/sync/builtin/db.go rename {dep/depsync => platform/dep/sync}/client.go (99%) rename {dep/depsync => platform/dep/sync}/depsync.go (74%) rename {dep/depsync => platform/dep/sync}/event.go (96%) rename {dep/depsync => platform/dep/sync}/get_autoassigners.go (73%) rename {dep/depsync => platform/dep/sync}/internal/depsyncproto/depsync.go (100%) rename {dep/depsync => platform/dep/sync}/internal/depsyncproto/depsync.pb.go (100%) rename {dep/depsync => platform/dep/sync}/internal/depsyncproto/depsync.proto (100%) rename {dep/depsync => platform/dep/sync}/remove_autoassigners.go (91%) rename {dep/depsync => platform/dep/sync}/server.go (94%) create mode 100644 platform/dep/sync/service.go rename {dep/depsync => platform/dep/sync}/syncnow.go (98%) diff --git a/cmd/mdmctl/apply_dep_autoassigner.go b/cmd/mdmctl/apply_dep_autoassigner.go index 3158093d..ac13d573 100644 --- a/cmd/mdmctl/apply_dep_autoassigner.go +++ b/cmd/mdmctl/apply_dep_autoassigner.go @@ -5,7 +5,7 @@ import ( "flag" "fmt" - "github.com/micromdm/micromdm/dep/depsync" + "github.com/micromdm/micromdm/platform/dep/sync" "github.com/pkg/errors" ) @@ -24,7 +24,7 @@ func (cmd *applyCommand) applyDEPAutoAssigner(args []string) error { return errors.New("bad input: must provide both -filter and -uuid") } - assigner := depsync.AutoAssigner{*flFilter, *flProfileUUID} + assigner := sync.AutoAssigner{*flFilter, *flProfileUUID} err := cmd.depsyncsvc.ApplyAutoAssigner(context.TODO(), &assigner) if err != nil { diff --git a/cmd/mdmctl/apply_dep_profile.go b/cmd/mdmctl/apply_dep_profile.go index e08fe122..82c0f874 100644 --- a/cmd/mdmctl/apply_dep_profile.go +++ b/cmd/mdmctl/apply_dep_profile.go @@ -13,8 +13,8 @@ import ( "github.com/pkg/errors" "github.com/micromdm/micromdm/dep" - "github.com/micromdm/micromdm/dep/depsync" "github.com/micromdm/micromdm/pkg/crypto" + "github.com/micromdm/micromdm/platform/dep/sync" ) func certificatesFromURL(serverURL string, insecure bool) ([]*x509.Certificate, error) { @@ -101,7 +101,7 @@ func (cmd *applyCommand) applyDEPProfile(args []string) error { fmt.Printf("Defined DEP Profile with UUID %s\n", resp.ProfileUUID) if *flFilter != "" { - assigner := depsync.AutoAssigner{*flFilter, resp.ProfileUUID} + assigner := sync.AutoAssigner{*flFilter, resp.ProfileUUID} err := cmd.depsyncsvc.ApplyAutoAssigner(context.TODO(), &assigner) if err != nil { return errors.Wrap(err, "set auto-assigner") diff --git a/cmd/mdmctl/setup.go b/cmd/mdmctl/setup.go index 3539c1b1..27422f78 100644 --- a/cmd/mdmctl/setup.go +++ b/cmd/mdmctl/setup.go @@ -4,11 +4,11 @@ import ( "github.com/go-kit/kit/log" httptransport "github.com/go-kit/kit/transport/http" - "github.com/micromdm/micromdm/dep/depsync" "github.com/micromdm/micromdm/platform/appstore" "github.com/micromdm/micromdm/platform/blueprint" "github.com/micromdm/micromdm/platform/config" "github.com/micromdm/micromdm/platform/dep" + "github.com/micromdm/micromdm/platform/dep/sync" "github.com/micromdm/micromdm/platform/device" "github.com/micromdm/micromdm/platform/profile" "github.com/micromdm/micromdm/platform/remove" @@ -24,7 +24,7 @@ type remoteServices struct { configsvc config.Service appsvc appstore.Service depsvc dep.Service - depsyncsvc depsync.Service + depsyncsvc sync.Service } func setupClient(logger log.Logger) (*remoteServices, error) { @@ -89,7 +89,7 @@ func setupClient(logger log.Logger) (*remoteServices, error) { return nil, err } - depsyncsvc, err := depsync.NewHTTPClient( + depsyncsvc, err := sync.NewHTTPClient( cfg.ServerURL, cfg.APIToken, logger, httptransport.SetClient(skipVerifyHTTPClient(cfg.SkipVerify))) if err != nil { diff --git a/cmd/micromdm/serve.go b/cmd/micromdm/serve.go index e2383208..43d430f1 100644 --- a/cmd/micromdm/serve.go +++ b/cmd/micromdm/serve.go @@ -29,7 +29,6 @@ import ( "github.com/pkg/errors" "golang.org/x/crypto/acme/autocert" - "github.com/micromdm/micromdm/dep/depsync" "github.com/micromdm/micromdm/mdm" "github.com/micromdm/micromdm/mdm/enroll" httputil2 "github.com/micromdm/micromdm/pkg/httputil" @@ -41,6 +40,7 @@ import ( "github.com/micromdm/micromdm/platform/command" "github.com/micromdm/micromdm/platform/config" depapi "github.com/micromdm/micromdm/platform/dep" + "github.com/micromdm/micromdm/platform/dep/sync" "github.com/micromdm/micromdm/platform/device" devicebuiltin "github.com/micromdm/micromdm/platform/device/builtin" "github.com/micromdm/micromdm/platform/profile" @@ -253,8 +253,8 @@ func serve(args []string) error { depEndpoints := depapi.MakeServerEndpoints(depsvc, basicAuthEndpointMiddleware) depapi.RegisterHTTPHandlers(r, depEndpoints, options...) - depsyncEndpoints := depsync.MakeServerEndpoints(depsync.NewService(syncer), basicAuthEndpointMiddleware) - depsync.RegisterHTTPHandlers(r, depsyncEndpoints, options...) + depsyncEndpoints := sync.MakeServerEndpoints(sync.NewService(syncer, sm.SyncDB), basicAuthEndpointMiddleware) + sync.RegisterHTTPHandlers(r, depsyncEndpoints, options...) } else { mainLogger.Log("msg", "no api key specified") } diff --git a/dep/depsync/config.go b/dep/depsync/config.go deleted file mode 100644 index eed2a7fe..00000000 --- a/dep/depsync/config.go +++ /dev/null @@ -1,95 +0,0 @@ -package depsync - -import ( - "encoding/json" - - "github.com/boltdb/bolt" - "github.com/pkg/errors" -) - -type config struct { - *bolt.DB - Cursor cursor `json:"cursor"` -} - -func (cfg *config) Save() error { - err := cfg.Update(func(tx *bolt.Tx) error { - bkt, err := tx.CreateBucketIfNotExists([]byte(ConfigBucket)) - if err != nil { - return err - } - v, err := json.Marshal(cfg) - if err != nil { - return err - } - return bkt.Put([]byte("configuration"), v) - }) - return errors.Wrap(err, "saving dep sync cursor") -} - -func (cfg *config) saveAutoAssigner(assigner *AutoAssigner) error { - if assigner.Filter != "*" { - return errors.New("only '*' filter auto-assigners supported") - } - err := cfg.Update(func(tx *bolt.Tx) error { - b, err := tx.CreateBucketIfNotExists([]byte(AutoAssignBucket)) - if err != nil { - return err - } - return b.Put([]byte(assigner.Filter), []byte(assigner.ProfileUUID)) - }) - return errors.Wrap(err, "saving auto-assigner") -} - -func (cfg *config) loadAutoAssigners() ([]*AutoAssigner, error) { - assigners := []*AutoAssigner{} - err := cfg.View(func(tx *bolt.Tx) error { - b := tx.Bucket([]byte(AutoAssignBucket)) - if b == nil { // bucket doesn't exist yet - return nil - } - - return b.ForEach(func(k, v []byte) error { - assigners = append(assigners, &AutoAssigner{ - Filter: string(k), - ProfileUUID: string(v), - }) - return nil - }) - }) - return assigners, errors.Wrap(err, "loading auto-assigners") -} - -func (cfg *config) deleteAutoAssigner(filter string) error { - return cfg.Update(func(tx *bolt.Tx) error { - b := tx.Bucket([]byte(AutoAssignBucket)) - if b == nil { // bucket doesn't exist yet - return nil - } - return b.Delete([]byte(filter)) - }) -} - -func LoadConfig(db *bolt.DB) (*config, error) { - conf := config{DB: db} - err := db.Update(func(tx *bolt.Tx) error { - bkt, err := tx.CreateBucketIfNotExists([]byte(ConfigBucket)) - if err != nil { - return err - } - - v := bkt.Get([]byte("configuration")) - if v == nil { - return nil - } - if err := json.Unmarshal(v, &conf); err != nil { - return err - } - return nil - }) - if err != nil { - return nil, err - } - - return &conf, nil -} diff --git a/dep/depsync/service.go b/dep/depsync/service.go deleted file mode 100644 index c669c906..00000000 --- a/dep/depsync/service.go +++ /dev/null @@ -1,16 +0,0 @@ -package depsync - -import ( - "context" -) - -type Service interface { - SyncNow(context.Context) error - ApplyAutoAssigner(context.Context, *AutoAssigner) error - GetAutoAssigners(context.Context) ([]*AutoAssigner, error) - RemoveAutoAssigner(context.Context, string) error -} - -type DEPSyncService struct { - syncer Syncer -} diff --git a/go.sum b/go.sum index b543fbe6..f94719ad 100644 --- a/go.sum +++ b/go.sum @@ -22,6 +22,7 @@ github.com/gorilla/mux v1.6.2 h1:Pgr17XVTNXAk3q/r4CpKzC5xBM/qW1uVLV+IhRZpIIk= github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs= github.com/groob/finalizer v0.0.0-20170707115354-4c2ed49aabda h1:5ikpG9mYCMFiZX0nkxoV6aU2IpCHPdws3gCNgdZeEV0= github.com/groob/finalizer v0.0.0-20170707115354-4c2ed49aabda/go.mod h1:MyndkAZd5rUMdNogn35MWXBX1UiBigrU8eTj8DoAC2c= +github.com/groob/pkcs7 v0.0.0-20180824154052-36585635cb64 h1:1ALD84dEnUxPKZENhUAeQ0tuJ+s3PuL85pV95B0Ekfk= github.com/groob/pkcs7 v0.0.0-20180824154052-36585635cb64/go.mod h1:mEOMQ8C7oeXY3LnE2jy4UkLAqrW9rrpwiP5U4hVV+MY= github.com/groob/plist v0.0.0-20180203051248-dd56909aee38 h1:afbUddvIjPRC7XHHgeSTRfzZtIxEsSl4VCxumLBGDJU= github.com/groob/plist v0.0.0-20180203051248-dd56909aee38/go.mod h1:qg2Nek0ND/hIr+nY8H1oVqEW2cLzVVNaAQ0QexOyjyc= diff --git a/dep/depsync/apply_autoassigner.go b/platform/dep/sync/apply_autoassigner.go similarity index 92% rename from dep/depsync/apply_autoassigner.go rename to platform/dep/sync/apply_autoassigner.go index fb9f236d..fd017e44 100644 --- a/dep/depsync/apply_autoassigner.go +++ b/platform/dep/sync/apply_autoassigner.go @@ -1,17 +1,18 @@ -package depsync +package sync import ( "context" "net/http" "github.com/go-kit/kit/endpoint" + "github.com/pkg/errors" "github.com/micromdm/micromdm/pkg/httputil" ) func (s DEPSyncService) ApplyAutoAssigner(ctx context.Context, aa *AutoAssigner) error { - conf := s.syncer.GetConfig() - return conf.saveAutoAssigner(aa) + err := s.db.SaveAutoAssigner(aa) + return errors.Wrap(err, "saving AutoAssigner") } type applyAutoAssignerRequest struct { diff --git a/platform/dep/sync/builtin/db.go b/platform/dep/sync/builtin/db.go new file mode 100644 index 00000000..10a0e0a7 --- /dev/null +++ b/platform/dep/sync/builtin/db.go @@ -0,0 +1,116 @@ +package builtin + +import ( + "encoding/json" + + "github.com/boltdb/bolt" + "github.com/pkg/errors" + + "github.com/micromdm/micromdm/platform/dep/sync" +) + +const ( + ConfigBucket = "mdm.DEPConfig" + AutoAssignBucket = "mdm.DEPAutoAssign" +) + +type DB struct { + *bolt.DB +} + +func NewDB(db *bolt.DB) (*DB, error) { + err := db.Update(func(tx *bolt.Tx) error { + _, err := tx.CreateBucketIfNotExists([]byte(ConfigBucket)) + if err != nil { + return err + } + _, err = tx.CreateBucketIfNotExists([]byte(AutoAssignBucket)) + return err + }) + if err != nil { + return nil, errors.Wrapf(err, "creating %s bucket", ConfigBucket) + } + datastore := &DB{DB: db} + return datastore, nil +} + +func (db *DB) LoadCursor() (*sync.Cursor, error) { + var cursor = struct { + Cursor sync.Cursor `json:"cursor"` + }{} + err := db.View(func(tx *bolt.Tx) error { + bkt := tx.Bucket([]byte(ConfigBucket)) + v := bkt.Get([]byte("configuration")) + if v == nil { + return nil // TODO add notfound + } + err := json.Unmarshal(v, &cursor) + return errors.Wrap(err, "unmarshal dep cursor") + }) + if err != nil { + return nil, errors.Wrap(err, "load cursor from bolt") + } + return &cursor.Cursor, nil +} + +func (db *DB) SaveCursor(c sync.Cursor) error { + err := db.Update(func(tx *bolt.Tx) error { + bkt, err := tx.CreateBucketIfNotExists([]byte(ConfigBucket)) + if err != nil { + return err + } + // use anonymous struct to preserve old structure. + var cursor = struct { + Cursor sync.Cursor `json:"cursor"` + }{Cursor: c} + v, err := json.Marshal(&cursor) + if err != nil { + return err + } + return bkt.Put([]byte("configuration"), v) + }) + return errors.Wrap(err, "saving dep sync cursor") +} + +func (db *DB) SaveAutoAssigner(a *sync.AutoAssigner) error { + if a.Filter != "*" { + return errors.New("only '*' filter auto-assigners supported") + } + err := db.Update(func(tx *bolt.Tx) error { + b, err := tx.CreateBucketIfNotExists([]byte(AutoAssignBucket)) + if err != nil { + return err + } + return b.Put([]byte(a.Filter), []byte(a.ProfileUUID)) + }) + return errors.Wrap(err, "saving auto-assigner") +} + +func (db *DB) DeleteAutoAssigner(filter string) error { + return db.Update(func(tx *bolt.Tx) error { + b := tx.Bucket([]byte(AutoAssignBucket)) + if b == nil { // bucket doesn't exist yet + return nil + } + return b.Delete([]byte(filter)) + }) +} + +func (db *DB) LoadAutoAssigners() ([]sync.AutoAssigner, error) { + var aa []sync.AutoAssigner + err := db.View(func(tx *bolt.Tx) error { + b := tx.Bucket([]byte(AutoAssignBucket)) + if b == nil { // bucket doesn't exist yet + return nil + } + + return b.ForEach(func(k, v []byte) error { + aa = append(aa, sync.AutoAssigner{ + Filter: string(k), + ProfileUUID: string(v), + }) + return nil + }) + }) + return aa, errors.Wrap(err, "loading auto-assigners") +} diff --git a/dep/depsync/client.go b/platform/dep/sync/client.go similarity index 99% rename from dep/depsync/client.go rename to platform/dep/sync/client.go index 038a7af8..e7a6e862 100644 --- a/dep/depsync/client.go +++ b/platform/dep/sync/client.go @@ -1,4 +1,4 @@ -package depsync +package sync import ( "net/url" diff --git a/dep/depsync/depsync.go b/platform/dep/sync/depsync.go similarity index 74% rename from dep/depsync/depsync.go rename to platform/dep/sync/depsync.go index 2a59299c..9f2362de 100644 --- a/dep/depsync/depsync.go +++ b/platform/dep/sync/depsync.go @@ -1,4 +1,4 @@ -package depsync +package sync import ( "context" @@ -7,7 +7,6 @@ import ( "sync" "time" - "github.com/boltdb/bolt" "github.com/go-kit/kit/log" "github.com/go-kit/kit/log/level" "github.com/pkg/errors" @@ -18,47 +17,81 @@ import ( ) const ( - SyncTopic = "mdm.DepSync" - ConfigBucket = "mdm.DEPConfig" - AutoAssignBucket = "mdm.DEPAutoAssign" + SyncTopic = "mdm.DepSync" syncDuration = 30 * time.Minute cursorValidDuration = 7 * 24 * time.Hour ) -type Syncer interface { - SyncNow() - GetConfig() *config // TODO: #302 +type Syncer interface{ SyncNow() } + +type WatcherDB interface { + LoadCursor() (*Cursor, error) + SaveCursor(c Cursor) error + LoadAutoAssigners() ([]AutoAssigner, error) } -type AutoAssigner struct { - Filter string `json:"filter"` - ProfileUUID string `json:"profile_uuid"` -} - -type watcher struct { +type Watcher struct { mtx sync.RWMutex logger log.Logger client Client publisher pubsub.Publisher - conf *config + db WatcherDB startSync chan bool syncNow chan bool + + cursor Cursor } -type cursor struct { - Value string `json:"value"` - CreatedAt time.Time `json:"created_at"` -} - -// A cursor is valid for a week. -func (c cursor) Valid() bool { - expiration := time.Now().Add(cursorValidDuration) - if c.CreatedAt.After(expiration) { - return false +func NewWatcher(db WatcherDB, pub pubsub.PublishSubscriber, opts ...Option) (*Watcher, error) { + w := Watcher{ + logger: log.NewNopLogger(), + db: db, + publisher: pub, + startSync: make(chan bool), + syncNow: make(chan bool), } - return true + for _, optFn := range opts { + optFn(&w) + } + + cursor, err := w.db.LoadCursor() + if err != nil { + return nil, err + } + if cursor.Valid() { + level.Debug(w.logger).Log("msg", "loaded DEP config", "cursor", cursor.Value) + w.cursor = *cursor + } + + if err := w.updateClient(pub); err != nil { + return nil, err + } + + saveCursor := func() { + if err := db.SaveCursor(w.cursor); err != nil { + level.Info(w.logger).Log("err", err, "msg", "saving cursor") + return + } + level.Info(w.logger).Log("msg", "saved DEP config", "cursor", w.cursor.Value) + } + + go func() { + defer saveCursor() + if w.client == nil { + panic("boom") + // block until we have a DEP client to start sync process + level.Info(w.logger).Log("msg", "waiting for DEP token to be added before starting sync") + <-w.startSync + } + err := w.Run() + // the DEP sync should never end without an error, but log + // unconditionally anyway so we never silently stop watching + level.Info(w.logger).Log("err", err, "msg", "DEP watcher stopped") + }() + + return &w, nil } type Client interface { @@ -67,71 +100,21 @@ type Client interface { AssignProfile(string, ...string) (*dep.ProfileResponse, error) } -type Option func(*watcher) +type Option func(*Watcher) func WithClient(client Client) Option { - return func(w *watcher) { + return func(w *Watcher) { w.client = client } } func WithLogger(logger log.Logger) Option { - return func(w *watcher) { + return func(w *Watcher) { w.logger = logger } } -func New(pub pubsub.PublishSubscriber, db *bolt.DB, logger log.Logger, opts ...Option) (Syncer, error) { - conf, err := LoadConfig(db) - if err != nil { - return nil, err - } - if conf.Cursor.Valid() { - level.Info(logger).Log("msg", "loaded DEP config", "cursor", conf.Cursor.Value) - } else { - conf.Cursor.Value = "" - } - - sync := &watcher{ - logger: log.NewNopLogger(), - publisher: pub, - conf: conf, - startSync: make(chan bool), - syncNow: make(chan bool), - } - - for _, opt := range opts { - opt(sync) - } - - if err := sync.updateClient(pub); err != nil { - return nil, err - } - - saveCursor := func() { - if err := conf.Save(); err != nil { - level.Info(logger).Log("err", err, "msg", "saving cursor") - return - } - level.Info(logger).Log("msg", "saved DEP config", "cursor", conf.Cursor.Value) - } - - go func() { - defer saveCursor() - if sync.client == nil { - // block until we have a DEP client to start sync process - level.Info(logger).Log("msg", "waiting for DEP token to be added before starting sync") - <-sync.startSync - } - err := sync.Run() - // the DEP sync should never end without an error, but log - // unconditionally anyway so we never silently stop watching - level.Info(logger).Log("err", err, "msg", "DEP watcher stopped") - }() - return sync, nil -} - -func (w *watcher) updateClient(pubsub pubsub.Subscriber) error { +func (w *Watcher) updateClient(pubsub pubsub.Subscriber) error { tokenAdded, err := pubsub.Subscribe(context.TODO(), "token-events", conf.DEPTokenTopic) if err != nil { return err @@ -163,7 +146,7 @@ func (w *watcher) updateClient(pubsub pubsub.Subscriber) error { return nil } -func (w *watcher) SyncNow() { +func (w *Watcher) SyncNow() { if w.client == nil { level.Info(w.logger).Log("msg", "waiting for DEP token to be added before starting sync") return @@ -171,10 +154,6 @@ func (w *watcher) SyncNow() { w.syncNow <- true } -func (w *watcher) GetConfig() *config { - return w.conf -} - // TODO this needs to be a proper error in the micromdm/dep package. func isCursorExhausted(err error) bool { return strings.Contains(err.Error(), "EXHAUSTED_CURSOR") @@ -190,12 +169,12 @@ func isCursorInvalid(err error) bool { // Process DEP messages and pull out filter-matching serial numbers // associated to profile UUIDs for auto-assignment. -func (w *watcher) filteredAutoAssignments(devices []dep.Device) (map[string][]string, error) { +func (w *Watcher) filteredAutoAssignments(devices []dep.Device) (map[string][]string, error) { // load auto-assigners every run to make sure we get the latest set of // auto-assigner profile UUIDs/filters. Note this makes every *watcher // (i.e. every DEP sync instance) share the current DB set of auto- // assigners. perhaps to refactor to be more separated. - assigners, err := w.conf.loadAutoAssigners() + assigners, err := w.db.LoadAutoAssigners() if err != nil { return nil, err } @@ -225,7 +204,7 @@ func (w *watcher) filteredAutoAssignments(devices []dep.Device) (map[string][]st return assigned, nil } -func (w *watcher) processAutoAssign(devices []dep.Device) error { +func (w *Watcher) processAutoAssign(devices []dep.Device) error { assignments, err := w.filteredAutoAssignments(devices) if err != nil { return err @@ -267,7 +246,7 @@ func (w *watcher) processAutoAssign(devices []dep.Device) error { return nil } -func (w *watcher) publishAndProcessDevices(devices []dep.Device) error { +func (w *Watcher) publishAndProcessDevices(devices []dep.Device) error { e := NewEvent(devices) data, err := MarshalEvent(e) if err != nil { @@ -291,21 +270,21 @@ func (w *watcher) publishAndProcessDevices(devices []dep.Device) error { return nil } -func (w *watcher) Run() error { +func (w *Watcher) Run() error { ticker := time.NewTicker(syncDuration).C FETCH: for { - resp, err := w.client.FetchDevices(dep.Limit(100), dep.Cursor(w.conf.Cursor.Value)) + resp, err := w.client.FetchDevices(dep.Limit(100), dep.Cursor(w.cursor.Value)) if err != nil && isCursorExhausted(err) { goto SYNC } else if err != nil && isCursorInvalid(err) { level.Info(w.logger).Log( "msg", "DEP fetch cursor response", - "cursor", w.conf.Cursor.Value, + "cursor", w.cursor.Value, "err", err, "msg", "retrying DEP fetch with empty cursor", ) - w.conf.Cursor.Value = "" + w.cursor.Value = "" goto FETCH } else if err != nil { return err @@ -317,8 +296,8 @@ FETCH: "fetched", resp.FetchedUntil, "devices", len(resp.Devices), ) - w.conf.Cursor = cursor{Value: resp.Cursor, CreatedAt: time.Now()} - if err := w.conf.Save(); err != nil { + w.cursor = Cursor{Value: resp.Cursor, CreatedAt: time.Now()} + if err := w.db.SaveCursor(w.cursor); err != nil { return errors.Wrap(err, "saving cursor from fetch") } if err := w.publishAndProcessDevices(resp.Devices); err != nil { @@ -331,15 +310,15 @@ FETCH: SYNC: for { - resp, err := w.client.SyncDevices(w.conf.Cursor.Value, dep.Cursor(w.conf.Cursor.Value)) + resp, err := w.client.SyncDevices(w.cursor.Value, dep.Cursor(w.cursor.Value)) if err != nil && (isCursorExpired(err) || isCursorInvalid(err)) { level.Info(w.logger).Log( "msg", "DEP sync cursor response", - "cursor", w.conf.Cursor.Value, + "cursor", w.cursor.Value, "err", err, "msg", "retrying DEP fetch with empty cursor", ) - w.conf.Cursor.Value = "" + w.cursor.Value = "" goto FETCH } else if err != nil { return err @@ -351,8 +330,8 @@ SYNC: "fetched", resp.FetchedUntil, "devices", len(resp.Devices), ) - w.conf.Cursor = cursor{Value: resp.Cursor, CreatedAt: time.Now()} - if err := w.conf.Save(); err != nil { + w.cursor = Cursor{Value: resp.Cursor, CreatedAt: time.Now()} + if err := w.db.SaveCursor(w.cursor); err != nil { return errors.Wrap(err, "saving cursor from sync") } if err := w.publishAndProcessDevices(resp.Devices); err != nil { diff --git a/dep/depsync/event.go b/platform/dep/sync/event.go similarity index 96% rename from dep/depsync/event.go rename to platform/dep/sync/event.go index 0792e85f..61a6b8fc 100644 --- a/dep/depsync/event.go +++ b/platform/dep/sync/event.go @@ -1,4 +1,4 @@ -package depsync +package sync import ( "time" @@ -7,7 +7,7 @@ import ( uuid "github.com/satori/go.uuid" "github.com/micromdm/micromdm/dep" - "github.com/micromdm/micromdm/dep/depsync/internal/depsyncproto" + "github.com/micromdm/micromdm/platform/dep/sync/internal/depsyncproto" ) type Event struct { diff --git a/dep/depsync/get_autoassigners.go b/platform/dep/sync/get_autoassigners.go similarity index 73% rename from dep/depsync/get_autoassigners.go rename to platform/dep/sync/get_autoassigners.go index 0f193b57..63f7f7f5 100644 --- a/dep/depsync/get_autoassigners.go +++ b/platform/dep/sync/get_autoassigners.go @@ -1,22 +1,23 @@ -package depsync +package sync import ( "context" "net/http" "github.com/go-kit/kit/endpoint" + "github.com/pkg/errors" "github.com/micromdm/micromdm/pkg/httputil" ) -func (s DEPSyncService) GetAutoAssigners(ctx context.Context) ([]*AutoAssigner, error) { - conf := s.syncer.GetConfig() - return conf.loadAutoAssigners() +func (s DEPSyncService) GetAutoAssigners(ctx context.Context) ([]AutoAssigner, error) { + aa, err := s.db.LoadAutoAssigners() + return aa, errors.Wrap(err, "get AutoAssigners") } type getAutoAssignersResponse struct { - AutoAssigners []*AutoAssigner `json:"autoassigners"` - Err error `json:"err,omitempty"` + AutoAssigners []AutoAssigner `json:"autoassigners"` + Err error `json:"err,omitempty"` } func (r getAutoAssignersResponse) Failed() error { return r.Err } @@ -37,7 +38,7 @@ func decodeGetAutoAssignersResponse(ctx context.Context, r *http.Response) (inte return req, err } -func (e Endpoints) GetAutoAssigners(ctx context.Context) ([]*AutoAssigner, error) { +func (e Endpoints) GetAutoAssigners(ctx context.Context) ([]AutoAssigner, error) { resp, err := e.GetAutoAssignersEndpoint(ctx, nil) if err != nil { return nil, err diff --git a/dep/depsync/internal/depsyncproto/depsync.go b/platform/dep/sync/internal/depsyncproto/depsync.go similarity index 100% rename from dep/depsync/internal/depsyncproto/depsync.go rename to platform/dep/sync/internal/depsyncproto/depsync.go diff --git a/dep/depsync/internal/depsyncproto/depsync.pb.go b/platform/dep/sync/internal/depsyncproto/depsync.pb.go similarity index 100% rename from dep/depsync/internal/depsyncproto/depsync.pb.go rename to platform/dep/sync/internal/depsyncproto/depsync.pb.go diff --git a/dep/depsync/internal/depsyncproto/depsync.proto b/platform/dep/sync/internal/depsyncproto/depsync.proto similarity index 100% rename from dep/depsync/internal/depsyncproto/depsync.proto rename to platform/dep/sync/internal/depsyncproto/depsync.proto diff --git a/dep/depsync/remove_autoassigners.go b/platform/dep/sync/remove_autoassigners.go similarity index 91% rename from dep/depsync/remove_autoassigners.go rename to platform/dep/sync/remove_autoassigners.go index db060871..be132a89 100644 --- a/dep/depsync/remove_autoassigners.go +++ b/platform/dep/sync/remove_autoassigners.go @@ -1,4 +1,4 @@ -package depsync +package sync import ( "context" @@ -6,11 +6,12 @@ import ( "github.com/go-kit/kit/endpoint" "github.com/micromdm/micromdm/pkg/httputil" + "github.com/pkg/errors" ) func (s DEPSyncService) RemoveAutoAssigner(ctx context.Context, filter string) error { - conf := s.syncer.GetConfig() - return conf.deleteAutoAssigner(filter) + err := s.db.DeleteAutoAssigner(filter) + return errors.Wrap(err, "remove AutoAssigner") } type removeAutoAssignerRequest struct { diff --git a/dep/depsync/server.go b/platform/dep/sync/server.go similarity index 94% rename from dep/depsync/server.go rename to platform/dep/sync/server.go index 1d56ccbe..fd220f02 100644 --- a/dep/depsync/server.go +++ b/platform/dep/sync/server.go @@ -1,4 +1,4 @@ -package depsync +package sync import ( "github.com/go-kit/kit/endpoint" @@ -8,8 +8,8 @@ import ( "github.com/micromdm/micromdm/pkg/httputil" ) -func NewService(syncer Syncer) *DEPSyncService { - return &DEPSyncService{syncer: syncer} +func NewService(syncer Syncer, db DB) *DEPSyncService { + return &DEPSyncService{syncer: syncer, db: db} } type Endpoints struct { diff --git a/platform/dep/sync/service.go b/platform/dep/sync/service.go new file mode 100644 index 00000000..7ce3bdcb --- /dev/null +++ b/platform/dep/sync/service.go @@ -0,0 +1,40 @@ +package sync + +import ( + "context" + "time" +) + +type Service interface { + SyncNow(context.Context) error + ApplyAutoAssigner(context.Context, *AutoAssigner) error + GetAutoAssigners(context.Context) ([]AutoAssigner, error) + RemoveAutoAssigner(context.Context, string) error +} + +type DB interface { + SaveAutoAssigner(a *AutoAssigner) error + LoadAutoAssigners() ([]AutoAssigner, error) + DeleteAutoAssigner(filter string) error +} + +type DEPSyncService struct { + db DB + syncer Syncer +} + +type Cursor struct { + Value string `json:"value"` + CreatedAt time.Time `json:"created_at"` +} + +// A cursor is valid for a week. +func (c Cursor) Valid() bool { + expiration := time.Now().Add(cursorValidDuration) + return c.CreatedAt.Before(expiration) +} + +type AutoAssigner struct { + Filter string `json:"filter"` + ProfileUUID string `json:"profile_uuid"` +} diff --git a/dep/depsync/syncnow.go b/platform/dep/sync/syncnow.go similarity index 98% rename from dep/depsync/syncnow.go rename to platform/dep/sync/syncnow.go index 288533d4..5a88a325 100644 --- a/dep/depsync/syncnow.go +++ b/platform/dep/sync/syncnow.go @@ -1,4 +1,4 @@ -package depsync +package sync import ( "context" diff --git a/platform/device/worker.go b/platform/device/worker.go index 51527b48..d14eb3f4 100644 --- a/platform/device/worker.go +++ b/platform/device/worker.go @@ -9,8 +9,8 @@ import ( "github.com/pkg/errors" uuid "github.com/satori/go.uuid" - "github.com/micromdm/micromdm/dep/depsync" "github.com/micromdm/micromdm/mdm" + "github.com/micromdm/micromdm/platform/dep/sync" "github.com/micromdm/micromdm/platform/pubsub" ) @@ -48,9 +48,9 @@ func (w *Worker) Run(ctx context.Context) error { if err != nil { return errors.Wrapf(err, "subscribing %s to %s", subscription, mdm.CheckoutTopic) } - depSyncEvents, err := w.ps.Subscribe(ctx, subscription, depsync.SyncTopic) + depSyncEvents, err := w.ps.Subscribe(ctx, subscription, sync.SyncTopic) if err != nil { - return errors.Wrapf(err, "subscribing %s to %s", subscription, depsync.SyncTopic) + return errors.Wrapf(err, "subscribing %s to %s", subscription, sync.SyncTopic) } connectEvents, err := w.ps.Subscribe(ctx, subscription, mdm.ConnectTopic) if err != nil { @@ -84,8 +84,8 @@ func (w *Worker) Run(ctx context.Context) error { } func (w *Worker) updateFromDEPSync(ctx context.Context, message []byte) error { - var ev depsync.Event - if err := depsync.UnmarshalEvent(message, &ev); err != nil { + var ev sync.Event + if err := sync.UnmarshalEvent(message, &ev); err != nil { return errors.Wrap(err, "unmarshal depsync event") } level.Debug(w.logger).Log( diff --git a/server/server.go b/server/server.go index 3b520fdd..32e7baeb 100644 --- a/server/server.go +++ b/server/server.go @@ -23,7 +23,6 @@ import ( "golang.org/x/crypto/pkcs12" "github.com/micromdm/micromdm/dep" - "github.com/micromdm/micromdm/dep/depsync" "github.com/micromdm/micromdm/mdm" "github.com/micromdm/micromdm/mdm/enroll" "github.com/micromdm/micromdm/pkg/crypto" @@ -32,6 +31,8 @@ import ( "github.com/micromdm/micromdm/platform/command" "github.com/micromdm/micromdm/platform/config" configbuiltin "github.com/micromdm/micromdm/platform/config/builtin" + "github.com/micromdm/micromdm/platform/dep/sync" + syncbuiltin "github.com/micromdm/micromdm/platform/dep/sync/builtin" "github.com/micromdm/micromdm/platform/profile" profilebuiltin "github.com/micromdm/micromdm/platform/profile/builtin" "github.com/micromdm/micromdm/platform/pubsub" @@ -60,6 +61,7 @@ type Server struct { RemoveDB block.Store CommandWebhookURL string DEPClient *dep.Client + SyncDB *syncbuiltin.DB PushService *push.Service // bufford push APNSPushService apns.Service @@ -398,17 +400,23 @@ func (c *Server) setupDepClient() error { return nil } -func (c *Server) CreateDEPSyncer(logger log.Logger) (depsync.Syncer, error) { +func (c *Server) CreateDEPSyncer(logger log.Logger) (sync.Syncer, error) { client := c.DEPClient - opts := []depsync.Option{ - depsync.WithLogger(log.With(logger, "component", "depsync")), + opts := []sync.Option{ + sync.WithLogger(log.With(logger, "component", "depsync")), } if client != nil { - opts = append(opts, depsync.WithClient(client)) + opts = append(opts, sync.WithClient(client)) } - var syncer depsync.Syncer - syncer, err := depsync.New(c.PubClient, c.DB, logger, opts...) + syncdb, err := syncbuiltin.NewDB(c.DB) + if err != nil { + return nil, err + } + c.SyncDB = syncdb + + var syncer sync.Syncer + syncer, err = sync.NewWatcher(c.SyncDB, c.PubClient, opts...) if err != nil { return nil, err }