Osquerybeat: emit pack_name and query_name in scheduled query results

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 {
@@ -190,7 +190,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 @@
 	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
@@ -59,7 +67,7 @@
 }
 
 func TestActionHandlerExecute(t *testing.T) {
 	validLogger := logp.NewLogger("action_test")
 	inputType := osqueryInputType
 
 	ctx := context.Background()
@@ -156,12 +164,12 @@
 				if ok {
 					t.Fatal("Unexpected error:", errVal)
 				} else if tc.Skipped {
 					diff := cmp.Diff("", tc.QueryExecutor.(*mockExecutor).receivedSql)
 					if diff != "" {
 						t.Error(diff)
 					}
 
 					diff = cmp.Diff("", tc.Publisher.(*mockPublisher).idValue)
 					if diff != "" {
 						t.Error(diff)
 					}
@@ -171,7 +179,7 @@
 						t.Error(diff)
 					}
 				} else {
 					diff := cmp.Diff(tc.QueryExecutor.(*mockExecutor).receivedSql, actionSQL)
 					if diff != "" {
 						t.Error(diff)
 					}

x-pack/osquerybeat/beater/config_plugin.go

@@ -49,6 +49,10 @@ type QueryInfo struct {
 	Interval int
 	// PackID is the policy-defined pack identifier for pack queries; empty for top-level schedule queries.
 	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
 }
@@ -293,8 +297,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 {
@@ -309,6 +315,8 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 			SpaceID:    qi.SpaceID,
 			Interval:   qi.Interval,
 			PackID:     packID,
+			PackName:   packName,
+			QueryName:  queryName,
 			Profile:    config.ResolveProfiling(globalProfile, qi.Profiling),
 		}
 		namespaces[name] = ns
@@ -327,7 +335,7 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 		if err := config.ValidateQueryScheduleMode(qi); err != nil {
 			return fmt.Errorf("osquery.schedule[%q]: %w", name, err)
 		}
-		qi, err = registerQuery(name, p.namespace, qi, "")
+		qi, err = registerQuery(name, p.namespace, qi, "", "", name)
 		if err != nil {
 			return err
 		}
@@ -348,7 +356,7 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 			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)
+			qi, err = registerQuery(getPackQueryName(packName, name), p.namespace, qi, packID, pack.PackName, name)
 			if err != nil {
 				return err
 			}
@@ -379,7 +387,7 @@ func (p *ConfigPlugin) set(inputs []config.InputConfig) (err error) {
 				Profiling:  stream.Profiling,
 			}
 
-			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

@@ -53,7 +53,7 @@
 }
 
 func TestConfigPluginNew(t *testing.T) {
 	validLogger := logp.NewLogger("config_test")
 
 	tests := []struct {
 		name        string
@@ -239,7 +239,7 @@
 }
 
 func TestSet(t *testing.T) {
 	logger := logp.NewLogger("config_test")
 
 	const noQueriesConfig = `{
     "options": {
@@ -635,6 +635,14 @@
 	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) {
@@ -644,6 +652,7 @@
 	const (
 		packName    = "my-pack"
 		packID      = "pack-uuid-123"
+		packLabel   = "My Pack"
 		queryName   = "uptime_query"
 		querySQL    = "select * from uptime"
 		queryPeriod = 60
@@ -659,7 +668,8 @@
 			Osquery: &config.OsqueryConfig{
 				Packs: map[string]config.Pack{
 					packName: {
-						PackID: packID,
+						PackID:   packID,
+						PackName: packLabel,
 						Queries: map[string]config.Query{
 							queryName: {
 								Query: querySQL,
@@ -692,6 +702,13 @@
 	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)
+	}
 }
 
 func TestSet_PackNativeMetadataWithoutIntervalRejected(t *testing.T) {

x-pack/osquerybeat/beater/osquerybeat.go

@@ -616,7 +616,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" {
@@ -654,7 +654,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{}) {

x-pack/osquerybeat/beater/recurrence_query_handler.go

@@ -321,11 +321,13 @@ func (h *recurrenceQueryHandler) executeQuery(ctx context.Context, scheduledQuer
 
 	// Get query info for ECS mapping, pack/space, and schedule id fallback
 	var ecsMapping ecs.Mapping
-	var spaceID, packID string
+	var spaceID, packID, packName, queryName string
 	if qi, ok := h.configPlugin.LookupQueryInfo(name); ok {
 		ecsMapping = qi.ECSMapping
 		spaceID = qi.SpaceID
 		packID = qi.PackID
+		packName = qi.PackName
+		queryName = qi.QueryName
 	}
 
 	scheduleID := scheduledQuery.ScheduleID()
@@ -349,7 +351,7 @@ func (h *recurrenceQueryHandler) executeQuery(ctx context.Context, scheduledQuer
 		if action != "" {
 			meta["action"] = action
 		}
-		h.publisher.Publish(config.Datastream(ns), scheduleID, "schedule_id", responseID, spaceID, packID, meta, rows, ecsMapping, nil)
+		h.publisher.Publish(config.Datastream(ns), scheduleID, "schedule_id", responseID, spaceID, packID, packName, queryName, meta, rows, ecsMapping, nil)
 	}
 
 	if scheduledQuery.Snapshot() {
@@ -361,7 +363,7 @@ func (h *recurrenceQueryHandler) executeQuery(ctx context.Context, scheduledQuer
 	}
 
 	// Synthetic response document (no action) with execution count for correlation
-	h.publisher.PublishScheduledResponse(scheduleID, packID, spaceID, responseID, startedAt, completedAt, plannedScheduleTime, totalHits, int64(executionIndex))
+	h.publisher.PublishScheduledResponse(scheduleID, packID, packName, queryName, spaceID, responseID, startedAt, completedAt, plannedScheduleTime, totalHits, int64(executionIndex))
 
 	h.log.Debugf("RRULE-scheduled query '%s' completed with %d results", name, len(hits))
 	return nil

x-pack/osquerybeat/internal/config/osquery.go

@@ -203,7 +203,10 @@ type Query struct {
 type Pack struct {
 	// PackID is the policy-defined pack identifier; used in result/response documents for correlation.
 	// If empty, the pack map key (pack name) is used when publishing.
-	PackID    string   `config:"pack_id,omitempty" json:"pack_id,omitempty"`
+	PackID string `config:"pack_id,omitempty" json:"pack_id,omitempty"`
+	// PackName is the policy-defined human-readable pack name; emitted as pack_name in
+	// result/response documents alongside pack_id. Optional and not sent to osqueryd.
+	PackName  string   `config:"pack_name,omitempty" json:"-"`
 	Discovery []string `config:"discovery" json:"discovery,omitempty"`
 	Platform  string   `config:"platform" json:"platform,omitempty"`
 	Version   string   `config:"version" json:"version,omitempty"`

x-pack/osquerybeat/internal/pub/publisher.go

@@ -153,12 +153,12 @@ func (p *Publisher) Configure(inputs []config.InputConfig) error {
 	return nil
 }
 
-func (p *Publisher) Publish(index, idValue, idFieldKey, responseID, spaceID, packID string, meta map[string]interface{}, hits []map[string]interface{}, ecsm ecs.Mapping, reqData interface{}) {
+func (p *Publisher) Publish(index, idValue, idFieldKey, responseID, spaceID, packID, packName, queryName string, meta map[string]interface{}, hits []map[string]interface{}, ecsm ecs.Mapping, reqData interface{}) {
 	p.mx.Lock()
 	defer p.mx.Unlock()
 
 	for _, hit := range hits {
-		event := hitToEvent(index, p.b.Info.Name, idValue, idFieldKey, responseID, spaceID, packID, meta, hit, ecsm, reqData)
+		event := hitToEvent(index, p.b.Info.Name, idValue, idFieldKey, responseID, spaceID, packID, packName, queryName, meta, hit, ecsm, reqData)
 		p.client.Publish(event)
 	}
 	p.log.Infof("%d events sent to index %s", len(hits), index)
@@ -201,7 +201,7 @@ func (p *Publisher) PublishActionResult(req map[string]interface{}, res map[stri
 // PublishScheduledResponse publishes a synthetic response document for a scheduled query run (no action).
 // Includes schedule_execution_count;
 // native uses 1 + (run_time - start_date) / interval).
-func (p *Publisher) PublishScheduledResponse(scheduleID, packID, spaceID, responseID string, startedAt, completedAt, plannedScheduleTime time.Time, resultCount int, scheduleExecutionCount int64) {
+func (p *Publisher) PublishScheduledResponse(scheduleID, packID, packName, queryName, spaceID, responseID string, startedAt, completedAt, plannedScheduleTime time.Time, resultCount int, scheduleExecutionCount int64) {
 	p.mx.Lock()
 	defer p.mx.Unlock()
 
@@ -227,6 +227,12 @@ func (p *Publisher) PublishScheduledResponse(scheduleID, packID, spaceID, respon
 	if packID != "" {
 		fields["pack_id"] = packID
 	}
+	if packName != "" {
+		fields["pack_name"] = packName
+	}
+	if queryName != "" {
+		fields["query_name"] = queryName
+	}
 	if spaceID != "" {
 		fields["space_id"] = spaceID
 	}
@@ -365,7 +371,7 @@ func (p *Publisher) processorsForInputConfig(inCfg config.InputConfig, defaultDa
 	return procs, nil
 }
 
-func hitToEvent(index, eventType, idValue, idFieldKey, responseID, spaceID, packID string, meta, hit map[string]interface{}, ecsm ecs.Mapping, reqData interface{}) beat.Event {
+func hitToEvent(index, eventType, idValue, idFieldKey, responseID, spaceID, packID, packName, queryName string, meta, hit map[string]interface{}, ecsm ecs.Mapping, reqData interface{}) beat.Event {
 	var fields mapstr.M
 
 	if len(ecsm) > 0 {
@@ -415,6 +421,12 @@ func hitToEvent(index, eventType, idValue, idFieldKey, responseID, spaceID, pack
 	if packID != "" {
 		event.Fields["pack_id"] = packID
 	}
+	if packName != "" {
+		event.Fields["pack_name"] = packName
+	}
+	if queryName != "" {
+		event.Fields["query_name"] = queryName
+	}
 	if index != "" {
 		event.Meta = mapstr.M{events.FieldMetaRawIndex: index}
 	}