Add action and state stores migrations

internal/pkg/agent/application/paths/files.go

@@ -23,13 +23,20 @@ const defaultAgentFleetFile = "fleet.enc"
 // defaultAgentEnrollFile is a name of file used to enroll agent on first-start
 const defaultAgentEnrollFile = "enroll.yml"
 
-// defaultAgentActionStoreFile is the file that will contain the action that can be replayed after restart.
+// defaultAgentActionStoreFile is the file that will contain the action that can
+// be replayed after restart.
+// It's deprecated and kept for migration purposes.
+// Deprecated.
 const defaultAgentActionStoreFile = "action_store.yml"
 
-// defaultAgentStateStoreYmlFile is the file that will contain the action that can be replayed after restart.
+// defaultAgentStateStoreYmlFile is the file that will contain the action that
+// can be replayed after restart.
+// It's deprecated and kept for migration purposes.
+// Deprecated.
 const defaultAgentStateStoreYmlFile = "state.yml"
 
-// defaultAgentStateStoreFile is the file that will contain the action that can be replayed after restart encrypted.
+// defaultAgentStateStoreFile is the file that will contain the encrypted state
+// store.
 const defaultAgentStateStoreFile = "state.enc"
 
 // defaultInputDPath return the location of the inputs.d.

internal/pkg/agent/storage/encrypted_disk_storage_windows_linux_test.go

@@ -109,4 +109,10 @@ func TestEncryptedDiskStorageWindowsLinuxLoad(t *testing.T) {
 	if diff != "" {
 		t.Error(diff)
 	}
+
+	// Save something else
+	err = s.Save(bytes.NewBuffer([]byte("new data")))
+	if err != nil {
+		t.Fatal(err)
+	}
 }

internal/pkg/agent/storage/encrypted_disk_store.go

@@ -88,8 +88,13 @@ func (d *EncryptedDiskStore) ensureKey(ctx context.Context) error {
 	return nil
 }
 
-// Save will write the encrypted storage to disk.
-// Specifically it will write to a .tmp file then rotate the file to the target name to ensure that an error does not corrupt the previously written file.
+// Save will read 'in' and write its contents encrypted to disk.
+// If EncryptedDiskStore.Load() was called, the io.ReadCloser it returns MUST be
+// closed before Save() can be called. It is so because Save() writes to a .tmp
+// file then rotate the file to the target name to ensure that an error does not
+// corrupt the previously written file.
+// Specially on windows systems, if the original files is still open because of
+// Load(), Save() would fail.
 func (d *EncryptedDiskStore) Save(in io.Reader) error {
 	// Ensure has agent key
 	err := d.ensureKey(d.ctx)

internal/pkg/agent/storage/storage.go

@@ -14,7 +14,9 @@ const perms os.FileMode = 0600
 
 // Store saves the io.Reader.
 type Store interface {
-	// Save the io.Reader.
+	// Save the io.Reader. Depending on the underlying implementation, if
+	// Storage.Load() was called, the io.ReadCloser MUST be closed before Save()
+	// can be called.
 	Save(io.Reader) error
 }
 

internal/pkg/agent/storage/store/internal/migrations/migrations.go

@@ -0,0 +1,95 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package migrations
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"time"
+
+	"gopkg.in/yaml.v2"
+)
+
+type StateStore struct {
+	Action      Action        `yaml:"action"`
+	AckToken    string        `yaml:"ack_token"`
+	ActionQueue []ActionQueue `yaml:"action_queue"`
+}
+
+// Action is a struct to read an Action Policy Change from its old
+// YAML format.
+type Action struct {
+	ActionID     string                 `yaml:"action_id"`
+	Type         string                 `yaml:"action_type"`
+	StartTime    time.Time              `yaml:"start_time,omitempty"`
+	SourceURI    string                 `yaml:"source_uri,omitempty"`
+	RetryAttempt int                    `yaml:"retry_attempt,omitempty"`
+	Policy       map[string]interface{} `yaml:"policy,omitempty"`
+	IsDetected   bool                   `yaml:"is_detected,omitempty"`
+}
+
+// ActionQueue is a struct to read the action queue from its old YAML format.
+type ActionQueue struct {
+	ActionID       string                 `yaml:"action_id"`
+	Type           string                 `yaml:"type"`
+	StartTime      time.Time              `yaml:"start_time,omitempty"`
+	ExpirationTime time.Time              `yaml:"expiration,omitempty"`
+	Version        string                 `yaml:"version,omitempty"`
+	SourceURI      string                 `yaml:"source_uri,omitempty"`
+	RetryAttempt   int                    `yaml:"retry_attempt,omitempty"`
+	Policy         map[string]interface{} `yaml:"policy,omitempty"`
+	IsDetected     bool                   `yaml:"is_detected,omitempty"`
+}
+
+type loader interface {
+	Load() (io.ReadCloser, error)
+}
+
+// LoadActionStore loads an action store from loader.
+func LoadActionStore(loader loader) (*Action, error) {
+	return LoadStore[Action](loader)
+}
+
+// LoadYAMLStateStore loads the old YAML state store from loader.
+func LoadYAMLStateStore(loader loader) (*StateStore, error) {
+	return LoadStore[StateStore](loader)
+}
+
+// LoadStore loads a YAML file.
+func LoadStore[Store any](loader loader) (store *Store, err error) {
+	// Store is a generic type, this might be needed.
+	store = new(Store)
+
+	reader, err := loader.Load()
+	if err != nil {
+		return nil, fmt.Errorf("failed to load action store: %w", err)
+	}
+	defer func() {
+		err2 := reader.Close()
+		if err != nil {
+			err = errors.Join(err,
+				fmt.Errorf("migration storeLoad failed to close reader: %w", err2))
+		}
+	}()
+
+	data, err := io.ReadAll(reader)
+	if err != nil {
+		return nil, err
+	}
+	buff := bytes.NewReader(data)
+
+	dec := yaml.NewDecoder(buff)
+	err = dec.Decode(&store)
+	if errors.Is(err, io.EOF) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("could not YAML unmarshal action from action store: %w", err)
+	}
+
+	return store, nil
+}