Add explicit DEP sync now (#404)

Fixes #175.

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) {

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")
+			}
 		}
 	}
 }

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{}

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}
+}

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
+}