From e32d2bea1b56aa121abf97953ab16131d7d01e39 Mon Sep 17 00:00:00 2001 From: Jesse Peterson Date: Sat, 5 May 2018 11:57:50 -0700 Subject: [PATCH] Add explicit DEP sync now (#404) Fixes #175. --- cmd/micromdm/serve.go | 17 ++++++++++++----- dep/depsync/depsync.go | 26 +++++++++++++++++++------- dep/depsync/endpoint.go | 20 ++++++++++++++++++++ dep/depsync/service.go | 22 ++++++++++++++++++++++ dep/depsync/transport_http.go | 31 +++++++++++++++++++++++++++++++ 5 files changed, 104 insertions(+), 12 deletions(-) create mode 100644 dep/depsync/endpoint.go create mode 100644 dep/depsync/service.go create mode 100644 dep/depsync/transport_http.go diff --git a/cmd/micromdm/serve.go b/cmd/micromdm/serve.go index fd1195a3..484bc028 100644 --- a/cmd/micromdm/serve.go +++ b/cmd/micromdm/serve.go @@ -161,7 +161,7 @@ func serve(args []string) error { sm.setupWebhooks() sm.setupCommandQueue(logger) sm.setupDepClient() - sm.setupDEPSync(logger) + syncer := sm.setupDEPSync(logger) if sm.err != nil { stdlog.Fatal(sm.err) } @@ -300,6 +300,10 @@ func serve(args []string) error { scepHandler := scep.ServiceHandler(ctx, sm.scepService, httpLogger) enrollHandlers := enroll.MakeHTTPHandlers(ctx, enroll.MakeServerEndpoints(sm.enrollService, sm.scepDepot), httptransport.ServerErrorLogger(httpLogger)) + + syncNowEndpoint := depsync.MakeSyncNowEndpoint(depsync.NewRPC(syncer)) + depsyncHandlers := depsync.MakeHTTPHandlers(ctx, depsync.Endpoints{SyncNowEndpoint: syncNowEndpoint}, connectOpts...) + r := mux.NewRouter() r.Handle("/version", version.Handler()) r.Handle("/mdm/checkin", mdmAuthSignMessageMiddleware(sm.scepDepot, checkinHandlers.CheckinHandler)).Methods("PUT") @@ -337,6 +341,7 @@ func serve(args []string) error { r.Handle("/v1/dep/devices", apiAuthMiddleware(*flAPIKey, depHandlers)) r.Handle("/v1/dep/account", apiAuthMiddleware(*flAPIKey, depHandlers)) r.Handle("/v1/dep/profiles", apiAuthMiddleware(*flAPIKey, depHandlers)) + r.Handle("/v1/dep/syncnow", apiAuthMiddleware(*flAPIKey, depsyncHandlers.SyncNowHandler)).Methods("POST") r.Handle("/v1/commands", apiAuthMiddleware(*flAPIKey, commandHandlers.NewCommandHandler)).Methods("POST") r.Handle("/push/{udid}", apiAuthMiddleware(*flAPIKey, apnsHandlers)) } else { @@ -769,9 +774,9 @@ func (c *server) setupDepClient() (dep.Client, error) { return client, nil } -func (c *server) setupDEPSync(logger log.Logger) { +func (c *server) setupDEPSync(logger log.Logger) depsync.Syncer { if c.err != nil { - return + return nil } client := c.depClient @@ -782,10 +787,12 @@ func (c *server) setupDEPSync(logger log.Logger) { opts = append(opts, depsync.WithClient(client)) } - _, c.err = depsync.New(c.pubclient, c.db, logger, opts...) + var syncer depsync.Syncer + syncer, c.err = depsync.New(c.pubclient, c.db, logger, opts...) if c.err != nil { - return + return nil } + return syncer } func (c *server) setupSCEP(logger log.Logger) { diff --git a/dep/depsync/depsync.go b/dep/depsync/depsync.go index f776841d..d48b1e00 100644 --- a/dep/depsync/depsync.go +++ b/dep/depsync/depsync.go @@ -20,10 +20,13 @@ import ( const ( SyncTopic = "mdm.DepSync" ConfigBucket = "mdm.DEPConfig" + + syncDuration = 30 * time.Minute + cursorValidDuration = 7 * 24 * time.Hour ) type Syncer interface { - privateDEPSyncer() bool + SyncNow() } type watcher struct { @@ -34,6 +37,7 @@ type watcher struct { publisher pubsub.Publisher conf *config startSync chan bool + syncNow chan bool } type cursor struct { @@ -43,7 +47,7 @@ type cursor struct { // A cursor is valid for a week. func (c cursor) Valid() bool { - expiration := time.Now().Add(24 * 7 * time.Hour) + expiration := time.Now().Add(cursorValidDuration) if c.CreatedAt.After(expiration) { return false } @@ -80,6 +84,7 @@ func New(pub pubsub.PublishSubscriber, db *bolt.DB, logger log.Logger, opts ...O publisher: pub, conf: conf, startSync: make(chan bool), + syncNow: make(chan bool), } for _, opt := range opts { @@ -144,9 +149,12 @@ func (w *watcher) updateClient(pubsub pubsub.Subscriber) error { return nil } -// TODO this is private temporarily until the interface can be defined -func (w *watcher) privateDEPSyncer() bool { - return true +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 + } + w.syncNow <- true } // TODO this needs to be a proper error in the micromdm/dep package. @@ -159,7 +167,7 @@ func isCursorExpired(err error) bool { } func (w *watcher) Run() error { - ticker := time.NewTicker(30 * time.Minute).C + ticker := time.NewTicker(syncDuration).C FETCH: for { resp, err := w.client.FetchDevices(dep.Limit(100), dep.Cursor(w.conf.Cursor.Value)) @@ -213,7 +221,11 @@ SYNC: } } if !resp.MoreToFollow { - <-ticker + select { + case <-ticker: + case <-w.syncNow: + level.Info(w.logger).Log("msg", "explicit DEP sync requested") + } } } } diff --git a/dep/depsync/endpoint.go b/dep/depsync/endpoint.go new file mode 100644 index 00000000..466ffc03 --- /dev/null +++ b/dep/depsync/endpoint.go @@ -0,0 +1,20 @@ +package depsync + +import ( + "context" + + "github.com/go-kit/kit/endpoint" +) + +type Endpoints struct { + SyncNowEndpoint endpoint.Endpoint +} + +func MakeSyncNowEndpoint(s Service) endpoint.Endpoint { + return func(ctx context.Context, request interface{}) (interface{}, error) { + s.SyncNow(ctx) + return syncNowResponse{}, nil + } +} + +type syncNowResponse struct{} diff --git a/dep/depsync/service.go b/dep/depsync/service.go new file mode 100644 index 00000000..a75df404 --- /dev/null +++ b/dep/depsync/service.go @@ -0,0 +1,22 @@ +package depsync + +import ( + "context" +) + +type Service interface { + SyncNow(ctx context.Context) +} + +type syncNowService struct { + syncer Syncer +} + +func (s *syncNowService) SyncNow(_ context.Context) { + s.syncer.SyncNow() + return +} + +func NewRPC(syncer Syncer) *syncNowService { + return &syncNowService{syncer: syncer} +} diff --git a/dep/depsync/transport_http.go b/dep/depsync/transport_http.go new file mode 100644 index 00000000..6e830aa5 --- /dev/null +++ b/dep/depsync/transport_http.go @@ -0,0 +1,31 @@ +package depsync + +import ( + "context" + "net/http" + + httptransport "github.com/go-kit/kit/transport/http" +) + +type HTTPHandlers struct { + SyncNowHandler http.Handler +} + +func MakeHTTPHandlers(ctx context.Context, endpoints Endpoints, opts ...httptransport.ServerOption) HTTPHandlers { + return HTTPHandlers{ + SyncNowHandler: httptransport.NewServer( + endpoints.SyncNowEndpoint, + decodeEmptyRequest, + encodeEmptyResponse, + opts..., + ), + } +} + +func decodeEmptyRequest(ctx context.Context, r *http.Request) (interface{}, error) { + return nil, nil +} + +func encodeEmptyResponse(ctx context.Context, w http.ResponseWriter, response interface{}) error { + return nil +}