osquery_perf: Windows MDM push (#46777)
<!-- Add the related story/sub-task/bug number, like Resolves #123, or remove if NA --> **Related issue:** Resolves #46567 Note: Hide whitespace for better review ## Testing - [x] QA'd all new/changed functionality manually <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Server-triggered on-demand Windows MDM check-ins for immediate device syncs * Dynamic adjustment of the device polling interval based on server directives * Enhanced metrics: tracking and reporting of on-demand MDM synchronization sessions <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Victor Lyuboslavsky
+143
-70
@@ -463,6 +463,11 @@ type agent struct {
|
||||
macMDMClient *mdmtest.TestAppleMDMClient
|
||||
winMDMClient *mdmtest.TestWindowsMDMClient
|
||||
|
||||
// winMDMWake signals the Windows MDM loop to start an OMA-DM session on demand (in response to the server's
|
||||
// WindowsMDMSyncRequest notification), mirroring real fleetd waking the device. Buffered with capacity 1, sent
|
||||
// non-blocking, so coalesced wakes never block the orbit config loop. Non-nil only for Windows MDM agents.
|
||||
winMDMWake chan struct{}
|
||||
|
||||
// isEnrolledToMDM is true when the mdmDevice has enrolled.
|
||||
isEnrolledToMDM bool
|
||||
// isEnrolledToMDMMu protects isEnrolledToMDM.
|
||||
@@ -729,6 +734,11 @@ func newAgent(
|
||||
entraIDUserPrincipalName: fmt.Sprintf("fake-%s@example.com", randomString(5)),
|
||||
}
|
||||
|
||||
// Windows MDM agents can be woken on demand by the server, so give them a wake channel for the MDM loop.
|
||||
if winMDMClient != nil {
|
||||
agent.winMDMWake = make(chan struct{}, 1)
|
||||
}
|
||||
|
||||
// Initialize host identity client
|
||||
agent.hostIdentityClient = hostidentity.NewClient(hostidentity.Config{
|
||||
ServerAddress: serverAddress,
|
||||
@@ -1013,6 +1023,16 @@ func (a *agent) runOrbitLoop() {
|
||||
|
||||
orbitClient.TestNodeKey = *a.orbitNodeKey
|
||||
|
||||
// Simulated Windows MDM hosts advertise CapabilityWindowsMDMSync, like real Windows fleetd, so the server relaxes
|
||||
// their DMClient poll schedule (via a Replace on the poll node) and wakes them on demand via WindowsMDMSyncRequest
|
||||
// instead of relying on frequent polling. Real fleetd adds this at construction (GetOrbitClientCapabilities gates it
|
||||
// on GOOS=windows); osquery-perf simulates Windows on a non-Windows GOOS, so we add it here. Mutating the map directly
|
||||
// is safe: GetOrbitClientCapabilities returns a fresh per-client map (not shared), and this runs during setup before
|
||||
// the first request or any concurrent goroutine, so there is no copy needed.
|
||||
if a.winMDMClient != nil {
|
||||
orbitClient.ClientCapabilities[fleet.CapabilityWindowsMDMSync] = struct{}{}
|
||||
}
|
||||
|
||||
deviceClient, err := fleetclient.NewDeviceClient(a.serverAddress, true, "", nil, "")
|
||||
if err != nil {
|
||||
log.Fatal("creating device client: ", err)
|
||||
@@ -1121,6 +1141,14 @@ func (a *agent) runOrbitLoop() {
|
||||
go a.runWindowsMDMLoop()
|
||||
}
|
||||
}
|
||||
if cfg.Notifications.WindowsMDMSyncRequest && a.mdmEnrolled() && a.winMDMWake != nil {
|
||||
// The server has queued Windows MDM commands and asked this (relaxed-poll) host to start an OMA-DM
|
||||
// session now.
|
||||
select {
|
||||
case a.winMDMWake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
case <-orbitTokenRemoteCheckTicker:
|
||||
if !a.disableFleetDesktop && tokenRotationEnabled {
|
||||
if err := deviceClient.CheckToken(*a.deviceAuthToken); err != nil {
|
||||
@@ -1526,83 +1554,128 @@ func (a *agent) ddmSendStatus(items *fleet.MDMAppleDDMDeclarationItemsResponse)
|
||||
}
|
||||
|
||||
func (a *agent) runWindowsMDMLoop() {
|
||||
mdmCheckInTicker := time.Tick(a.MDMCheckInInterval)
|
||||
pollInterval := a.MDMCheckInInterval
|
||||
ticker := time.NewTicker(pollInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for range mdmCheckInTicker {
|
||||
cmds, err := a.winMDMClient.StartManagementSession()
|
||||
if err != nil {
|
||||
log.Printf("MDM check-in start session request failed: %s", err)
|
||||
a.stats.IncrementMDMErrors()
|
||||
continue
|
||||
}
|
||||
a.stats.IncrementMDMSessions()
|
||||
|
||||
// send a successful ack for each command
|
||||
msgID, err := a.winMDMClient.GetCurrentMsgID()
|
||||
if err != nil {
|
||||
log.Printf("MDM get current MsgID failed: %s", err)
|
||||
a.stats.IncrementMDMErrors()
|
||||
continue
|
||||
}
|
||||
|
||||
// Detect SCEP CertificateInstall CSPs and ACK them now while kicking off the SCEP exchange in the background.
|
||||
scepCtx, cancelSCEP := context.WithTimeout(context.Background(), 2*a.MDMCheckInInterval)
|
||||
handled, scepResults, hasWork := a.winMDMClient.AppendSCEPInstallResponses(scepCtx, cmds, msgID, nil)
|
||||
if hasWork {
|
||||
go func() {
|
||||
defer cancelSCEP()
|
||||
// One SCEPResult is emitted per CSP; increment per-result so the request counter
|
||||
// matches the per-CSP success/error counters even when multiple CSPs ride one SyncML.
|
||||
for res := range scepResults {
|
||||
a.stats.IncrementMDMSCEPRequests()
|
||||
if res.Err != nil {
|
||||
log.Printf("MDM SCEP exchange failed: %s", res.Err)
|
||||
a.stats.IncrementMDMSCEPErrors()
|
||||
continue
|
||||
}
|
||||
a.stats.IncrementMDMSCEPSuccess()
|
||||
}
|
||||
}()
|
||||
} else {
|
||||
cancelSCEP()
|
||||
}
|
||||
|
||||
for _, c := range cmds {
|
||||
// Skip the server's own <Status> entries. MS-MDM's "Status on a Status" is only for auth-renegotiation edge cases (see
|
||||
// https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-mdm/36b1a4d9-fd93-48ce-b865-6a9d396c52a4
|
||||
// "While this case is not usually encountered"); real Windows does not emit Status-on-Status during normal check-ins.
|
||||
if c.Verb == fleet.CmdStatus {
|
||||
continue
|
||||
for {
|
||||
onDemand := false
|
||||
select {
|
||||
case <-a.winMDMWake:
|
||||
// The server asked us (via WindowsMDMSyncRequest) to start a session now, without waiting for the poll.
|
||||
onDemand = true
|
||||
case <-ticker.C:
|
||||
// A wake may have arrived at almost the same time as this tick; select picks at random when both are ready,
|
||||
// which could miscount a wake-triggered session as poll-triggered. Drain any buffered wake first so the
|
||||
// on-demand metric is deterministic (and a coincident tick+wake collapses into a single on-demand session).
|
||||
select {
|
||||
case <-a.winMDMWake:
|
||||
onDemand = true
|
||||
default:
|
||||
}
|
||||
if _, ok := handled[c.Cmd.CmdID.Value]; ok {
|
||||
// Already ACKed by AppendSCEPInstallResponses.
|
||||
a.stats.IncrementMDMCommandsReceived()
|
||||
continue
|
||||
}
|
||||
a.stats.IncrementMDMCommandsReceived()
|
||||
|
||||
status := syncml.CmdStatusOK
|
||||
if a.mdmProfileFailureProb > 0.0 && rand.Float64() <= a.mdmProfileFailureProb {
|
||||
status = syncml.CmdStatusBadRequest
|
||||
}
|
||||
a.winMDMClient.AppendResponse(fleet.SyncMLCmd{
|
||||
XMLName: xml.Name{Local: fleet.CmdStatus},
|
||||
MsgRef: &msgID,
|
||||
CmdRef: &c.Cmd.CmdID.Value,
|
||||
Cmd: ptr.String(c.Verb),
|
||||
Data: &status,
|
||||
Items: nil,
|
||||
CmdID: fleet.CmdID{Value: uuid.NewString()},
|
||||
})
|
||||
}
|
||||
if _, err := a.winMDMClient.SendResponse(); err != nil {
|
||||
log.Printf("MDM send response request failed: %s", err)
|
||||
a.stats.IncrementMDMErrors()
|
||||
continue
|
||||
|
||||
// If the server relaxed (or restored) our DMClient poll schedule via a Replace on the poll node, match it so our
|
||||
// scheduled polling slows down. Steady-state command delivery is then driven by the on-demand wake rather than
|
||||
// frequent polling, which is the behavior this load test exercises.
|
||||
if relaxedInterval := a.doWindowsMDMCheckIn(onDemand); relaxedInterval > 0 && relaxedInterval != pollInterval {
|
||||
pollInterval = relaxedInterval
|
||||
ticker.Reset(pollInterval)
|
||||
log.Printf("host %d: Windows MDM poll schedule set to %s", a.agentIndex, pollInterval)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// doWindowsMDMCheckIn runs a single OMA-DM management session: it starts the session, acknowledges every command the
|
||||
// server sends, and returns the new scheduled poll interval if the server sent a Replace on the DMClient poll node
|
||||
// (0 otherwise). onDemand reports whether the session was triggered by a WindowsMDMSyncRequest wake instead of the poll
|
||||
// ticker, for stats purposes.
|
||||
func (a *agent) doWindowsMDMCheckIn(onDemand bool) (newPollInterval time.Duration) {
|
||||
cmds, err := a.winMDMClient.StartManagementSession()
|
||||
if err != nil {
|
||||
log.Printf("MDM check-in start session request failed: %s", err)
|
||||
a.stats.IncrementMDMErrors()
|
||||
return 0
|
||||
}
|
||||
a.stats.IncrementMDMSessions()
|
||||
if onDemand {
|
||||
a.stats.IncrementMDMOnDemandSyncs()
|
||||
}
|
||||
|
||||
// send a successful ack for each command
|
||||
msgID, err := a.winMDMClient.GetCurrentMsgID()
|
||||
if err != nil {
|
||||
log.Printf("MDM get current MsgID failed: %s", err)
|
||||
a.stats.IncrementMDMErrors()
|
||||
return 0
|
||||
}
|
||||
|
||||
// Detect SCEP CertificateInstall CSPs and ACK them now while kicking off the SCEP exchange in the background.
|
||||
scepCtx, cancelSCEP := context.WithTimeout(context.Background(), 2*a.MDMCheckInInterval)
|
||||
handled, scepResults, hasWork := a.winMDMClient.AppendSCEPInstallResponses(scepCtx, cmds, msgID, nil)
|
||||
if hasWork {
|
||||
go func() {
|
||||
defer cancelSCEP()
|
||||
// One SCEPResult is emitted per CSP; increment per-result so the request counter
|
||||
// matches the per-CSP success/error counters even when multiple CSPs ride one SyncML.
|
||||
for res := range scepResults {
|
||||
a.stats.IncrementMDMSCEPRequests()
|
||||
if res.Err != nil {
|
||||
log.Printf("MDM SCEP exchange failed: %s", res.Err)
|
||||
a.stats.IncrementMDMSCEPErrors()
|
||||
continue
|
||||
}
|
||||
a.stats.IncrementMDMSCEPSuccess()
|
||||
}
|
||||
}()
|
||||
} else {
|
||||
cancelSCEP()
|
||||
}
|
||||
|
||||
for _, c := range cmds {
|
||||
// Skip the server's own <Status> entries. MS-MDM's "Status on a Status" is only for auth-renegotiation edge cases (see
|
||||
// https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-mdm/36b1a4d9-fd93-48ce-b865-6a9d396c52a4
|
||||
// "While this case is not usually encountered"); real Windows does not emit Status-on-Status during normal check-ins.
|
||||
if c.Verb == fleet.CmdStatus {
|
||||
continue
|
||||
}
|
||||
// When the server relaxes (or restores) our poll schedule it sends a Replace on the DMClient poll node. Honor it
|
||||
// by adjusting our scheduled poll interval, mirroring how the real Windows DMClient applies the Replace. We still
|
||||
// ACK it below like any other command.
|
||||
if c.Verb == fleet.CmdReplace && c.Cmd.GetTargetURI() == syncml.DMClientPollIntervalLocURI {
|
||||
if mins, err := strconv.Atoi(strings.TrimSpace(c.Cmd.GetTargetData())); err == nil && mins > 0 {
|
||||
newPollInterval = time.Duration(mins) * time.Minute
|
||||
}
|
||||
}
|
||||
if _, ok := handled[c.Cmd.CmdID.Value]; ok {
|
||||
// Already ACKed by AppendSCEPInstallResponses.
|
||||
a.stats.IncrementMDMCommandsReceived()
|
||||
continue
|
||||
}
|
||||
a.stats.IncrementMDMCommandsReceived()
|
||||
|
||||
status := syncml.CmdStatusOK
|
||||
if a.mdmProfileFailureProb > 0.0 && rand.Float64() <= a.mdmProfileFailureProb {
|
||||
status = syncml.CmdStatusBadRequest
|
||||
}
|
||||
a.winMDMClient.AppendResponse(fleet.SyncMLCmd{
|
||||
XMLName: xml.Name{Local: fleet.CmdStatus},
|
||||
MsgRef: &msgID,
|
||||
CmdRef: &c.Cmd.CmdID.Value,
|
||||
Cmd: ptr.String(c.Verb),
|
||||
Data: &status,
|
||||
Items: nil,
|
||||
CmdID: fleet.CmdID{Value: uuid.NewString()},
|
||||
})
|
||||
}
|
||||
if _, err := a.winMDMClient.SendResponse(); err != nil {
|
||||
log.Printf("MDM send response request failed: %s", err)
|
||||
a.stats.IncrementMDMErrors()
|
||||
return 0
|
||||
}
|
||||
return newPollInterval
|
||||
}
|
||||
|
||||
func (a *agent) execScripts(execIDs []string, orbitClient *fleetclient.OrbitClient) {
|
||||
if a.scriptExecRunning.Swap(true) {
|
||||
// if Swap returns true, the goroutine was already running, exit
|
||||
|
||||
@@ -13,6 +13,7 @@ type Stats struct {
|
||||
orbitEnrollments int
|
||||
mdmEnrollments int
|
||||
mdmSessions int
|
||||
mdmOnDemandSyncs int
|
||||
distributedWrites int
|
||||
mdmCommandsReceived int
|
||||
mdmSCEPRequests int
|
||||
@@ -77,6 +78,14 @@ func (s *Stats) IncrementMDMSessions() {
|
||||
s.mdmSessions++
|
||||
}
|
||||
|
||||
// IncrementMDMOnDemandSyncs counts Windows MDM sessions that were triggered by an on-demand wake
|
||||
// (WindowsMDMSyncRequest) rather than the poll ticker. This is a subset of mdmSessions, not a separate total.
|
||||
func (s *Stats) IncrementMDMOnDemandSyncs() {
|
||||
s.l.Lock()
|
||||
defer s.l.Unlock()
|
||||
s.mdmOnDemandSyncs++
|
||||
}
|
||||
|
||||
func (s *Stats) IncrementDistributedWrites() {
|
||||
s.l.Lock()
|
||||
defer s.l.Unlock()
|
||||
@@ -265,7 +274,7 @@ func (s *Stats) Log() {
|
||||
defer s.l.Unlock()
|
||||
|
||||
log.Printf(
|
||||
"uptime: %s, error rate: %.2f, osquery enrolls: %d, orbit enrolls: %d, mdm enrolls: %d, distributed/reads: %d, distributed/writes: %d, config requests: %d, result log requests: %d, mdm sessions initiated: %d, mdm commands received: %d, config errors: %d, distributed/read errors: %d, distributed/write errors: %d, log result errors: %d, orbit errors: %d, desktop errors: %d, mdm errors: %d, mdm scep requests: %d, mdm scep success: %d, mdm scep errors: %d, ddm tokens success: %d, ddm tokens errors: %d, ddm declaration items success: %d, ddm declaration items errors: %d, ddm activation success: %d, ddm activation errors: %d, ddm configuration success: %d, ddm configuration errors: %d, ddm status success: %d, ddm status errors: %d, buffered logs: %d, script execs (errs): %d (%d), software installs (errs): %d (%d)",
|
||||
"uptime: %s, error rate: %.2f, osquery enrolls: %d, orbit enrolls: %d, mdm enrolls: %d, distributed/reads: %d, distributed/writes: %d, config requests: %d, result log requests: %d, mdm sessions initiated: %d, mdm on-demand syncs: %d, mdm commands received: %d, config errors: %d, distributed/read errors: %d, distributed/write errors: %d, log result errors: %d, orbit errors: %d, desktop errors: %d, mdm errors: %d, mdm scep requests: %d, mdm scep success: %d, mdm scep errors: %d, ddm tokens success: %d, ddm tokens errors: %d, ddm declaration items success: %d, ddm declaration items errors: %d, ddm activation success: %d, ddm activation errors: %d, ddm configuration success: %d, ddm configuration errors: %d, ddm status success: %d, ddm status errors: %d, buffered logs: %d, script execs (errs): %d (%d), software installs (errs): %d (%d)",
|
||||
time.Since(s.StartTime).Round(time.Second),
|
||||
float64(s.errors)/float64(s.osqueryEnrollments),
|
||||
s.osqueryEnrollments,
|
||||
@@ -276,6 +285,7 @@ func (s *Stats) Log() {
|
||||
s.configRequests,
|
||||
s.resultLogRequests,
|
||||
s.mdmSessions,
|
||||
s.mdmOnDemandSyncs,
|
||||
s.mdmCommandsReceived,
|
||||
s.configErrors,
|
||||
s.distributedReadErrors,
|
||||
|
||||
Reference in New Issue
Block a user