Osquerybeat: emit pack_name and query_name in scheduled query results (#51781)

Scheduled pack queries (both osquery native interval schedules and osquerybeat
RRULE schedules) now include pack_name and query_name in their result and
response documents. pack_name comes from a new optional `pack_name` config
field on a pack (alongside the existing pack_id); query_name comes from the
pack's queries config map key. Both fields are only emitted when set.

These fields are required by the Osquery Manager dashboards, which group and
label scheduled query results by pack and query name.

(cherry picked from commit 5d82c71)

# Conflicts:
#	x-pack/osquerybeat/beater/config_plugin.go
#	x-pack/osquerybeat/beater/recurrence_query_handler.go
#	x-pack/osquerybeat/internal/config/osquery.go

Author: marc-gr

changelog/fragments/1783335960-osquerybeat-pack-name-query-name.yaml

@@ -0,0 +1,40 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: Emit pack_name and query_name in osquerybeat scheduled query results
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: |
+  Scheduled pack queries (both osquery native interval schedules and osquerybeat
+  RRULE schedules) now include two additional fields in their result and response
+  documents: pack_name, taken from the new optional `pack_name` config field on a
+  pack (alongside the existing pack_id), and query_name, taken from the pack's
+  queries config map key. Both fields are only emitted when set. These fields are
+  required by the Osquery Manager dashboards, which group and label scheduled
+  query results by pack and query name; without them the dashboards cannot render
+  those results correctly.
+
+# Affected component; a word indicating the component this changeset affects.
+component: osquerybeat
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+#pr: https://github.com/owner/repo/1234
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

x-pack/osquerybeat/beater/action_handler.go

@@ -26,11 +26,11 @@ type actionResultPublisher interface {
 }
 
 type queryResultPublisher interface {
-	Publish(index, idValue, idFieldKey, responseID, spaceID, packID string, meta map[string]interface{}, hits []map[string]interface{}, ecsm ecs.Mapping, reqData interface{})
+	Publish(index, idValue, idFieldKey, responseID, spaceID, packID, packName, queryName string, meta map[string]interface{}, hits []map[string]interface{}, ecsm ecs.Mapping, reqData interface{})
 }
 
 type scheduledResponsePublisher interface {
-	PublishScheduledResponse(scheduleID, packID, spaceID, responseID string, startedAt, completedAt, plannedScheduleTime time.Time, resultCount int, scheduleExecutionCount int64)
+	PublishScheduledResponse(scheduleID, packID, packName, queryName, spaceID, responseID string, startedAt, completedAt, plannedScheduleTime time.Time, resultCount int, scheduleExecutionCount int64)
 }
 
 type queryProfilePublisher interface {
@@ -172,7 +172,7 @@ func (a *actionHandler) executeQuery(ctx context.Context, index string, ac actio
 
 	a.log.Debugf("Completed query in: %v", duration)
 
-	a.publisher.Publish(index, ac.ID, "action_id", responseID, "", "", nil, hits, ac.ECSMapping, req["data"])
+	a.publisher.Publish(index, ac.ID, "action_id", responseID, "", "", "", "", nil, hits, ac.ECSMapping, req["data"])
 
 	return len(hits), nil
 }

x-pack/osquerybeat/beater/action_handler_test.go

@@ -36,18 +36,26 @@ type mockPublisher struct {
 	idValue    string
 	idFieldKey string
 	responseID string
+	spaceID    string
+	packID     string
+	packName   string
+	queryName  string
 	meta       map[string]interface{}
 	hits       []map[string]interface{}
 	ecsm       ecs.Mapping
 	reqData    interface{}
 	profile    map[string]interface{}
 }
 
-func (p *mockPublisher) Publish(index, idValue, idFieldKey, responseID, spaceID, packID string, meta map[string]interface{}, hits []map[string]interface{}, ecsm ecs.Mapping, reqData interface{}) {
+func (p *mockPublisher) Publish(index, idValue, idFieldKey, responseID, spaceID, packID, packName, queryName string, meta map[string]interface{}, hits []map[string]interface{}, ecsm ecs.Mapping, reqData interface{}) {
 	p.index = index
 	p.idValue = idValue
 	p.idFieldKey = idFieldKey
 	p.responseID = responseID
+	p.spaceID = spaceID
+	p.packID = packID
+	p.packName = packName
+	p.queryName = queryName
 	p.meta = meta
 	p.hits = hits
 	p.ecsm = ecsm

x-pack/osquerybeat/beater/config_plugin.go

@@ -43,8 +43,18 @@ type QueryInfo struct {
 	// Interval is the schedule interval in seconds for native schedules; used to compute schedule_execution_count
 	Interval int
 	// PackID is the policy-defined pack identifier for pack queries; empty for top-level schedule queries.
+<<<<<<< HEAD
 	PackID  string
 	Profile bool // whether to collect and publish profile for this query
+=======
+	PackID string
+	// PackName is the policy-defined human-readable pack name for pack queries; empty for top-level schedule queries.
+	PackName string
+	// QueryName is the query's config map key (pack query name or top-level schedule name).
+	QueryName string
+	// Profile is whether to collect and publish profile for this query.
+	Profile bool
+>>>>>>> 5d82c7117 (Osquerybeat: emit pack_name and query_name in scheduled query results (#51781))
 }
 
 type queryInfoMap map[string]QueryInfo
@@ -234,8 +244,10 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 		osqueryConfig = inputs[0].Osquery
 	}
 
-	// Common code to register query with lookup maps, enforce snapshot and increment queries count
-	registerQuery := func(name, ns string, qi config.Query, packID string) (config.Query, error) {
+	// Common code to register query with lookup maps, enforce snapshot and increment queries count.
+	// queryName is the config map key (pack query name or top-level schedule name); packID/packName
+	// are the pack identifiers for pack queries and empty for top-level schedule queries.
+	registerQuery := func(name, ns string, qi config.Query, packID, packName, queryName string) (config.Query, error) {
 		var ecsm ecs.Mapping
 		ecsm, err = flattenECSMapping(qi.ECSMapping)
 		if err != nil {
@@ -250,7 +262,13 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 			SpaceID:    qi.SpaceID,
 			Interval:   qi.Interval,
 			PackID:     packID,
+<<<<<<< HEAD
 			Profile:    qi.Profile,
+=======
+			PackName:   packName,
+			QueryName:  queryName,
+			Profile:    config.ResolveProfiling(globalProfile, qi.Profiling),
+>>>>>>> 5d82c7117 (Osquerybeat: emit pack_name and query_name in scheduled query results (#51781))
 		}
 		namespaces[name] = ns
 		queriesCount++
@@ -265,7 +283,14 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 
 	// Iterate osquery configuration's scheduled queries, add flattened ECS mappings to lookup map
 	for name, qi := range osqueryConfig.Schedule {
+<<<<<<< HEAD
 		qi, err = registerQuery(name, p.namespace, qi, "")
+=======
+		if err := config.ValidateQueryScheduleMode(qi); err != nil {
+			return fmt.Errorf("osquery.schedule[%q]: %w", name, err)
+		}
+		qi, err = registerQuery(name, p.namespace, qi, "", "", name)
+>>>>>>> 5d82c7117 (Osquerybeat: emit pack_name and query_name in scheduled query results (#51781))
 		if err != nil {
 			return err
 		}
@@ -279,7 +304,15 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 			packID = packName
 		}
 		for name, qi := range pack.Queries {
+<<<<<<< HEAD
 			qi, err = registerQuery(getPackQueryName(packName, name), p.namespace, qi, packID)
+=======
+			qi, err = config.MergeQueryWithPackScheduleDefaults(pack, qi)
+			if err != nil {
+				return fmt.Errorf("osquery.packs[%q].queries[%q]: %w", packName, name, err)
+			}
+			qi, err = registerQuery(getPackQueryName(packName, name), p.namespace, qi, packID, pack.PackName, name)
+>>>>>>> 5d82c7117 (Osquerybeat: emit pack_name and query_name in scheduled query results (#51781))
 			if err != nil {
 				return err
 			}
@@ -305,7 +338,7 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 				Profile:    stream.Profile,
 			}
 
-			qi, err = registerQuery(getPackQueryName(input.Name, stream.ID), p.namespace, qi, input.Name)
+			qi, err = registerQuery(getPackQueryName(input.Name, stream.ID), p.namespace, qi, input.Name, "", stream.ID)
 			if err != nil {
 				return err
 			}

x-pack/osquerybeat/beater/config_plugin_test.go

@@ -531,6 +531,14 @@ func TestSet_ScheduleMetadataIncludesSpaceID(t *testing.T) {
 	if diff := cmp.Diff(queryPeriod, qi.Interval); diff != "" {
 		t.Error(diff)
 	}
+	// query_name for a top-level scheduled query is the schedule config map key.
+	if diff := cmp.Diff(queryName, qi.QueryName); diff != "" {
+		t.Error(diff)
+	}
+	// Top-level schedule queries do not belong to a pack.
+	if diff := cmp.Diff("", qi.PackName); diff != "" {
+		t.Error(diff)
+	}
 }
 
 func TestSet_ScheduleMetadataIncludesPackID(t *testing.T) {
@@ -540,6 +548,7 @@ func TestSet_ScheduleMetadataIncludesPackID(t *testing.T) {
 	const (
 		packName    = "my-pack"
 		packID      = "pack-uuid-123"
+		packLabel   = "My Pack"
 		queryName   = "uptime_query"
 		querySQL    = "select * from uptime"
 		queryPeriod = 60
@@ -555,7 +564,8 @@ func TestSet_ScheduleMetadataIncludesPackID(t *testing.T) {
 			Osquery: &config.OsqueryConfig{
 				Packs: map[string]config.Pack{
 					packName: {
-						PackID: packID,
+						PackID:   packID,
+						PackName: packLabel,
 						Queries: map[string]config.Query{
 							queryName: {
 								Query:    querySQL,
@@ -586,4 +596,11 @@ func TestSet_ScheduleMetadataIncludesPackID(t *testing.T) {
 	if diff := cmp.Diff(packID, qi.PackID); diff != "" {
 		t.Error(diff)
 	}
+	if diff := cmp.Diff(packLabel, qi.PackName); diff != "" {
+		t.Error(diff)
+	}
+	// query_name is taken from the pack's queries config map key.
+	if diff := cmp.Diff(queryName, qi.QueryName); diff != "" {
+		t.Error(diff)
+	}
 }

x-pack/osquerybeat/beater/osquerybeat.go

@@ -592,7 +592,7 @@ func (bt *osquerybeat) handleQueryResult(ctx context.Context, cli *osqdcli.Clien
 	publishResolved := func(resultType, action string, hits []map[string]interface{}) {
 		totalHits += len(hits)
 		meta := queryResultMeta(resultType, action, res, scheduleExecutionCount, plannedScheduleTime)
-		bt.pub.Publish(config.Datastream(ns), scheduleID, "schedule_id", responseID, qi.SpaceID, qi.PackID, meta, hits, qi.ECSMapping, nil)
+		bt.pub.Publish(config.Datastream(ns), scheduleID, "schedule_id", responseID, qi.SpaceID, qi.PackID, qi.PackName, qi.QueryName, meta, hits, qi.ECSMapping, nil)
 	}
 
 	if res.Action == "snapshot" {
@@ -630,7 +630,7 @@ func (bt *osquerybeat) handleQueryResult(ctx context.Context, cli *osqdcli.Clien
 		}
 	}
 
-	bt.pub.PublishScheduledResponse(scheduleID, qi.PackID, qi.SpaceID, responseID, runTime, runTime, plannedScheduleTime, totalHits, scheduleExecutionCount)
+	bt.pub.PublishScheduledResponse(scheduleID, qi.PackID, qi.PackName, qi.QueryName, qi.SpaceID, responseID, runTime, runTime, plannedScheduleTime, totalHits, scheduleExecutionCount)
 }
 
 func queryResultMeta(typ, action string, res QueryResult, scheduleExecutionCount int64, plannedScheduleTime time.Time) map[string]interface{} {

x-pack/osquerybeat/beater/osquerybeat_publisher_test.go

@@ -21,13 +21,13 @@ type mockBeatPublisher struct {
 var _ osquerybeatPublisher = (*mockBeatPublisher)(nil)
 var _ scheduledQueryPublisher = (*mockBeatPublisher)(nil)
 
-func (m *mockBeatPublisher) Publish(index, idValue, idFieldKey, responseID, spaceID, packID string, meta map[string]interface{}, hits []map[string]interface{}, ecsm ecs.Mapping, reqData interface{}) {
+func (m *mockBeatPublisher) Publish(index, idValue, idFieldKey, responseID, spaceID, packID, packName, queryName string, meta map[string]interface{}, hits []map[string]interface{}, ecsm ecs.Mapping, reqData interface{}) {
 }
 
 func (m *mockBeatPublisher) PublishActionResult(req map[string]interface{}, res map[string]interface{}) {
 }
 
-func (m *mockBeatPublisher) PublishScheduledResponse(scheduleID, packID, spaceID, responseID string, startedAt, completedAt, plannedScheduleTime time.Time, resultCount int, scheduleExecutionCount int64) {
+func (m *mockBeatPublisher) PublishScheduledResponse(scheduleID, packID, packName, queryName, spaceID, responseID string, startedAt, completedAt, plannedScheduleTime time.Time, resultCount int, scheduleExecutionCount int64) {
 }
 
 func (m *mockBeatPublisher) PublishQueryProfile(index, queryName, actionID, responseID string, profile map[string]interface{}, reqData interface{}) {