[Elastic-Agent] Added source uri reloading (#686)

internal/pkg/agent/application/managed_mode.go

@@ -31,6 +31,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/agent/operation"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/storage"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/storage/store"
+	"github.com/elastic/elastic-agent/internal/pkg/artifact"
 	"github.com/elastic/elastic-agent/internal/pkg/capabilities"
 	"github.com/elastic/elastic-agent/internal/pkg/composable"
 	"github.com/elastic/elastic-agent/internal/pkg/config"
@@ -157,6 +158,7 @@ func newManaged(
 		},
 		caps,
 		monitor,
+		artifact.NewReloader(cfg.Settings.DownloadConfig, log),
 	)
 	if err != nil {
 		return nil, err

internal/pkg/agent/application/pipeline/actions/handlers/handler_action_policy_change.go

@@ -151,7 +151,7 @@ func (h *PolicyChange) handleFleetServerHosts(ctx context.Context, c *config.Con
 			errors.TypeNetwork, errors.M("hosts", h.config.Fleet.Client.Hosts))
 	}
 	// discard body for proper cancellation and connection reuse
-	io.Copy(ioutil.Discard, resp.Body)
+	_, _ = io.Copy(ioutil.Discard, resp.Body)
 	resp.Body.Close()
 
 	reader, err := fleetToReader(h.agentInfo, h.config)

internal/pkg/agent/application/pipeline/actions/handlers/handler_action_policy_change_test.go

@@ -44,7 +44,7 @@ func TestPolicyChange(t *testing.T) {
 
 		conf := map[string]interface{}{"hello": "world"}
 		action := &fleetapi.ActionPolicyChange{
-			ActionID:   "abc123",
+			ActionID:   "TestPolicyChange-abc1",
 			ActionType: "POLICY_CHANGE",
 			Policy:     conf,
 		}
@@ -69,7 +69,7 @@ func TestPolicyChange(t *testing.T) {
 
 		conf := map[string]interface{}{"hello": "world"}
 		action := &fleetapi.ActionPolicyChange{
-			ActionID:   "abc123",
+			ActionID:   "TestPolicyChange-abc2",
 			ActionType: "POLICY_CHANGE",
 			Policy:     conf,
 		}
@@ -100,7 +100,7 @@ func TestPolicyAcked(t *testing.T) {
 		emitter := &mockEmitter{err: mockErr}
 
 		config := map[string]interface{}{"hello": "world"}
-		actionID := "abc123"
+		actionID := "TestPolicyAcked-abc1"
 		action := &fleetapi.ActionPolicyChange{
 			ActionID:   actionID,
 			ActionType: "POLICY_CHANGE",
@@ -129,7 +129,7 @@ func TestPolicyAcked(t *testing.T) {
 		emitter := &mockEmitter{}
 
 		config := map[string]interface{}{"hello": "world"}
-		actionID := "abc123"
+		actionID := "TestPolicyAcked-abc2"
 		action := &fleetapi.ActionPolicyChange{
 			ActionID:   actionID,
 			ActionType: "POLICY_CHANGE",

internal/pkg/artifact/config.go

@@ -11,12 +11,17 @@ import (
 
 	"github.com/elastic/elastic-agent-libs/transport/httpcommon"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
+	"github.com/elastic/elastic-agent/internal/pkg/config"
+	"github.com/elastic/elastic-agent/pkg/core/logger"
 )
 
 const (
 	darwin  = "darwin"
 	linux   = "linux"
 	windows = "windows"
+
+	defaultSourceURI = "https://artifacts.elastic.co/downloads/"
 )
 
 // Config is a configuration used for verifier and downloader
@@ -46,6 +51,52 @@ type Config struct {
 	httpcommon.HTTPTransportSettings `config:",inline" yaml:",inline"` // Note: use anonymous struct for json inline
 }
 
+type Reloader struct {
+	log *logger.Logger
+	cfg *Config
+}
+
+func NewReloader(cfg *Config, log *logger.Logger) *Reloader {
+	return &Reloader{
+		cfg: cfg,
+		log: log,
+	}
+}
+
+func (r *Reloader) Reload(rawConfig *config.Config) error {
+	type reloadConfig struct {
+		// SourceURI: source of the artifacts, e.g https://artifacts.elastic.co/downloads/
+		SourceURI string `json:"agent.download.sourceURI" config:"agent.download.sourceURI"`
+
+		// FleetSourceURI: source of the artifacts, e.g https://artifacts.elastic.co/downloads/ coming from fleet which uses
+		// different naming.
+		FleetSourceURI string `json:"agent.download.source_uri" config:"agent.download.source_uri"`
+	}
+	cfg := &reloadConfig{}
+	if err := rawConfig.Unpack(&cfg); err != nil {
+		return errors.New(err, "failed to unpack config during reload")
+	}
+
+	var newSourceURI string
+	if cfg.FleetSourceURI != "" {
+		// fleet configuration takes precedence
+		newSourceURI = cfg.FleetSourceURI
+	} else if cfg.SourceURI != "" {
+		newSourceURI = cfg.SourceURI
+	}
+
+	if newSourceURI != "" {
+		r.log.Infof("Source URI changed from %q to %q", r.cfg.SourceURI, newSourceURI)
+		r.cfg.SourceURI = newSourceURI
+	} else {
+		// source uri unset, reset to default
+		r.log.Infof("Source URI reset from %q to %q", r.cfg.SourceURI, defaultSourceURI)
+		r.cfg.SourceURI = defaultSourceURI
+	}
+
+	return nil
+}
+
 // DefaultConfig creates a config with pre-set default values.
 func DefaultConfig() *Config {
 	transport := httpcommon.DefaultHTTPTransportSettings()
@@ -56,7 +107,7 @@ func DefaultConfig() *Config {
 	transport.Timeout = 10 * time.Minute
 
 	return &Config{
-		SourceURI:             "https://artifacts.elastic.co/downloads/",
+		SourceURI:             defaultSourceURI,
 		TargetDirectory:       paths.Downloads(),
 		InstallPath:           paths.Install(),
 		HTTPTransportSettings: transport,