Add top level keys for policy definition into specs (#3048)

Define agent policy structure as part of the openapi and json schema specs.
Use the generated json schema struct when fleet-server interacts with Elasticsearch.

changelog/fragments/1698276551-Define-policy-data-schema.yaml

@@ -0,0 +1,35 @@
+# 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: enhancement
+
+# Change summary; a 80ish characters long description of the change.
+summary: Define policy data schema
+
+# 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: |
+  Define the policy.Data section as part of the JSON schema used to interact with Elasticsearch.
+  All top-level properties have been kept generic as this is the first definition.
+  The only exception to this is the secret_references property which is fully defined.
+
+# Affected component; a word indicating the component this changeset affects.
+component:
+
+# 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

internal/pkg/api/handleCheckin.go

@@ -28,7 +28,6 @@ import (
 	"github.com/elastic/fleet-server/v7/internal/pkg/model"
 	"github.com/elastic/fleet-server/v7/internal/pkg/monitor"
 	"github.com/elastic/fleet-server/v7/internal/pkg/policy"
-	"github.com/elastic/fleet-server/v7/internal/pkg/smap"
 	"github.com/elastic/fleet-server/v7/internal/pkg/sqn"
 
 	"github.com/hashicorp/go-version"
@@ -685,54 +684,32 @@ func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, a
 		return nil, err
 	}
 
-	// Parse the outputs maps in order to prepare the outputs
-	const outputsProperty = "outputs"
-	outputs, err := smap.Parse(pp.Fields[outputsProperty])
-	if err != nil {
-		return nil, err
-	}
-
-	if outputs == nil {
+	if len(pp.Policy.Data.Outputs) == 0 {
 		return nil, ErrNoPolicyOutput
 	}
 
 	// Iterate through the policy outputs and prepare them
 	for _, policyOutput := range pp.Outputs {
-		err = policyOutput.Prepare(ctx, zlog, bulker, &agent, outputs)
+		err = policyOutput.Prepare(ctx, zlog, bulker, &agent, pp.Policy.Data.Outputs)
 		if err != nil {
 			return nil, fmt.Errorf("failed to prepare output %q:: %w",
 				policyOutput.Name, err)
 		}
 	}
+	// Add replace inputs with prepared version
+	pp.Policy.Data.Inputs = pp.Inputs
 
-	outputRaw, err := json.Marshal(outputs)
+	p, err := json.Marshal(pp.Policy.Data)
 	if err != nil {
 		return nil, err
 	}
 
-	// Dupe field map; pp is immutable
-	fields := make(map[string]json.RawMessage, len(pp.Fields))
-
-	for k, v := range pp.Fields {
-		fields[k] = v
-	}
-
-	// Update only the output fields to avoid duping the whole map
-	fields[outputsProperty] = json.RawMessage(outputRaw)
-
-	// replace agent policy inputs with the processed inputs where the secret references were replaced with the secret values
-	inputsRaw, err := json.Marshal(pp.Inputs)
-	if err != nil {
-		return nil, err
-	}
-
-	fields["inputs"] = json.RawMessage(inputsRaw)
-
 	rewrittenPolicy := struct {
-		Policy map[string]json.RawMessage `json:"policy"`
-	}{fields}
+		Policy json.RawMessage `json:"policy"`
+	}{p}
 
 	r := policy.RevisionFromPolicy(pp.Policy)
+	// FIXME use better action definition from open api spec
 	resp := Action{
 		AgentId:   agent.Id,
 		CreatedAt: pp.Policy.Timestamp,

internal/pkg/coordinator/monitor_integration_test.go

@@ -54,7 +54,7 @@ func TestMonitorLeadership(t *testing.T) {
 	policy1 := model.Policy{
 		PolicyID:       policy1Id,
 		CoordinatorIdx: 0,
-		Data:           []byte("{}"),
+		Data:           nil,
 		RevisionIdx:    1,
 	}
 	_, err = dl.CreatePolicy(ctx, bulker, policy1, dl.WithIndexName(policiesIndex))
@@ -85,7 +85,7 @@ func TestMonitorLeadership(t *testing.T) {
 	policy2 := model.Policy{
 		PolicyID:       policy2Id,
 		CoordinatorIdx: 0,
-		Data:           []byte("{}"),
+		Data:           nil,
 		RevisionIdx:    1,
 	}
 	_, err = dl.CreatePolicy(ctx, bulker, policy2, dl.WithIndexName(policiesIndex))

internal/pkg/coordinator/v0.go

@@ -6,7 +6,6 @@ package coordinator
 
 import (
 	"context"
-	"encoding/json"
 
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
@@ -73,13 +72,12 @@ func (c *coordinatorZeroT) Output() <-chan model.Policy {
 
 // updatePolicy performs the working of incrementing the coordinator idx.
 func (c *coordinatorZeroT) updatePolicy(p model.Policy) error {
-	newData, err := c.handlePolicy(p.Data)
+	_, err := c.handlePolicy(p.Data)
 	if err != nil {
 		return err
 	}
-	if p.CoordinatorIdx == 0 || string(newData) != string(p.Data) {
+	if p.CoordinatorIdx == 0 {
 		p.CoordinatorIdx += 1
-		p.Data = newData
 		c.policy = p
 		c.out <- p
 	}
@@ -89,6 +87,6 @@ func (c *coordinatorZeroT) updatePolicy(p model.Policy) error {
 // handlePolicy performs the actual work of coordination.
 //
 // Does nothing at the moment.
-func (c *coordinatorZeroT) handlePolicy(data json.RawMessage) (json.RawMessage, error) {
+func (c *coordinatorZeroT) handlePolicy(data *model.PolicyData) (*model.PolicyData, error) {
 	return data, nil
 }

internal/pkg/coordinator/v0_test.go

@@ -23,7 +23,7 @@ func TestCoordinatorZero(t *testing.T) {
 	policy := model.Policy{
 		PolicyID:       policyId,
 		CoordinatorIdx: 0,
-		Data:           []byte("{}"),
+		Data:           nil,
 		RevisionIdx:    1,
 	}
 	coord, err := NewCoordinatorZero(policy)
@@ -55,7 +55,7 @@ func TestCoordinatorZero(t *testing.T) {
 	policy = model.Policy{
 		PolicyID:       policyId,
 		CoordinatorIdx: 0,
-		Data:           []byte("{}"),
+		Data:           nil,
 		RevisionIdx:    2,
 	}
 	if err := coord.Update(ctx, policy); err != nil {
@@ -77,7 +77,7 @@ func TestCoordinatorZero(t *testing.T) {
 	policy = model.Policy{
 		PolicyID:       policyId,
 		CoordinatorIdx: 1,
-		Data:           []byte("{}"),
+		Data:           nil,
 		RevisionIdx:    2,
 	}
 	if err := coord.Update(ctx, policy); err != nil {

internal/pkg/dl/policies_integration_test.go

@@ -25,7 +25,7 @@ func createRandomPolicy(id string, revisionIdx int) model.Policy {
 		PolicyID:           id,
 		RevisionIdx:        int64(revisionIdx),
 		CoordinatorIdx:     0,
-		Data:               []byte("{}"),
+		Data:               nil,
 		DefaultFleetServer: false,
 		Timestamp:          now.Format(time.RFC3339),
 	}

internal/pkg/policy/monitor.go

@@ -173,7 +173,6 @@ LOOP:
 }
 
 func unmarshalHits(hits []es.HitT) ([]model.Policy, error) {
-
 	policies := make([]model.Policy, len(hits))
 	for i, hit := range hits {
 		err := hit.Unmarshal(&policies[i])