Cherry-pick #19507 to 7.x: Agent unenroll

go.sum

@@ -99,8 +99,10 @@ github.com/aerospike/aerospike-client-go v1.27.1-0.20170612174108-0f3b54da6bdc/g
 github.com/akavel/rsrc v0.8.0 h1:zjWn7ukO9Kc5Q62DOJCcxGpXC18RawVtYAGdz2aLlfw=
 github.com/akavel/rsrc v0.8.0/go.mod h1:uLoCtb9J+EyAqh+26kdrTgmzRBFPGOolLWKpdxkKq+c=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
+github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafoB+tBA3gMyHYHrpOtNuDiK/uB5uXxq5wM=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4 h1:Hs82Z41s6SdL1CELW+XaDYmOH4hkBN4/N9og/AsOv7E=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/andrewkroh/goja v0.0.0-20190128172624-dd2ac4456e20 h1:7rj9qZ63knnVo2ZeepYHvHuRdG76f3tRUTdIQDzRBeI=
 github.com/andrewkroh/goja v0.0.0-20190128172624-dd2ac4456e20/go.mod h1:cI59GRkC2FRaFYtgbYEqMlgnnfvAwXzjojyZKXwklNg=
@@ -880,6 +882,7 @@ google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzi
 google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
+gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

x-pack/elastic-agent/pkg/agent/application/action_store.go

@@ -7,6 +7,7 @@ package application
 import (
  "context"
  "fmt"
+ "io"
 
  yaml "gopkg.in/yaml.v2"
 
@@ -53,7 +54,7 @@ func newActionStore(log *logger.Logger, store storeLoad) (*actionStore, error) {
 // any other type of action will be silently ignored.
 func (s *actionStore) Add(a action) {
  switch v := a.(type) {
- case *fleetapi.ActionConfigChange:
+ case *fleetapi.ActionConfigChange, *fleetapi.ActionUnenroll:
  // Only persist the action if the action is different.
  if s.action != nil && s.action.ID() == v.ID() {
  return
@@ -69,16 +70,29 @@ func (s *actionStore) Save() error {
  return nil
  }
 
- apc, ok := s.action.(*fleetapi.ActionConfigChange)
- if !ok {
- return fmt.Errorf("incompatible type, expected ActionPolicyChange and received %T", s.action)
- }
+ var reader io.Reader
+ if apc, ok := s.action.(*fleetapi.ActionConfigChange); ok {
+ serialize := actionConfigChangeSerializer(*apc)
 
- serialize := actionConfigChangeSerializer(*apc)
+ r, err := yamlToReader(&serialize)
+ if err != nil {
+ return err
+ }
 
- reader, err := yamlToReader(&serialize)
- if err != nil {
- return err
+ reader = r
+ } else if aun, ok := s.action.(*fleetapi.ActionUnenroll); ok {
+ serialize := actionUnenrollSerializer(*aun)
+
+ r, err := yamlToReader(&serialize)
+ if err != nil {
+ return err
+ }
+
+ reader = r
+ }
+
+ if reader == nil {
+ return fmt.Errorf("incompatible type, expected ActionPolicyChange and received %T", s.action)
  }
 
  if err := s.store.Save(reader); err != nil {
@@ -98,7 +112,7 @@ func (s *actionStore) Actions() []action {
  return []action{s.action}
 }
 
-// actionConfigChangeSerializer is a struct that add YAML serialization, I don't think serialization
+// actionConfigChangeSerializer is a struct that adds a YAML serialization, I don't think serialization
 // is a concern of the fleetapi package. I went this route so I don't have to do much refactoring.
 //
 // There are four ways to achieve the same results:
@@ -117,6 +131,15 @@ type actionConfigChangeSerializer struct {
 // Add a guards between the serializer structs and the original struct.
 var _ actionConfigChangeSerializer = actionConfigChangeSerializer(fleetapi.ActionConfigChange{})
 
+// actionUnenrollSerializer is a struct that adds a YAML serialization,
+type actionUnenrollSerializer struct {
+ ActionID string `yaml:"action_id"`
+ ActionType string `yaml:"action_type"`
+}
+
+// Add a guards between the serializer structs and the original struct.
+var _ actionUnenrollSerializer = actionUnenrollSerializer(fleetapi.ActionUnenroll{})
+
 // actionStoreAcker wraps an existing acker and will send any acked event to the action store,
 // its up to the action store to decide if we need to persist the event for future replay or just
 // discard the event.

x-pack/elastic-agent/pkg/agent/application/handler_action_unenroll.go

@@ -0,0 +1,56 @@
+// 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 application
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/program"
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/fleetapi"
+)
+
+// After running Unenroll agent is in idle state, non managed non standalone.
+// For it to be operational again it needs to be either enrolled or reconfigured.
+type handlerUnenroll struct {
+ log *logger.Logger
+ emitter emitterFunc
+ dispatcher programsDispatcher
+ closers []context.CancelFunc
+}
+
+func (h *handlerUnenroll) Handle(ctx context.Context, a action, acker fleetAcker) error {
+ h.log.Debugf("handlerUnenroll: action '%+v' received", a)
+ action, ok := a.(*fleetapi.ActionUnenroll)
+ if !ok {
+ return fmt.Errorf("invalid type, expected ActionUnenroll and received %T", a)
+ }
+
+ // Providing empty map will close all pipelines
+ noPrograms := make(map[routingKey][]program.Program)
+ h.dispatcher.Dispatch(a.ID(), noPrograms)
+
+ if err := acker.Ack(ctx, action); err != nil {
+ return err
+ }
+
+ // commit all acks before quitting.
+ if err := acker.Commit(ctx); err != nil {
+ return err
+ }
+
+ // close fleet gateway loop
+ for _, c := range h.closers {
+ c()
+ }
+
+ // clean action store
+ // if err := os.Remove(info.AgentActionStoreFile()); err != nil && !os.IsNotExist(err) {
+ // return errors.New(err, "failed to clear action store")
+ // }
+
+ return nil
+}

x-pack/elastic-agent/pkg/agent/application/local_mode.go

@@ -26,6 +26,7 @@ type emitterFunc func(*config.Config) error
 // ConfigHandler is capable of handling config, perform actions at it, shutdown any long running process.
 type ConfigHandler interface {
  HandleConfig(configrequest.Request) error
+ Close() error
  Shutdown()
 }
 

x-pack/elastic-agent/pkg/agent/application/managed_mode.go

@@ -52,6 +52,7 @@ type Managed struct {
  gateway *fleetGateway
  router *router
  srv *server.Server
+ as *actionStore
 }
 
 func newManaged(
@@ -168,6 +169,7 @@ func newManaged(
  if err != nil {
  return nil, errors.New(err, fmt.Sprintf("fail to read action store '%s'", info.AgentActionStoreFile()))
  }
+ managedApplication.as = actionStore
  actionAcker := newActionStoreAcker(batchedAcker, actionStore)
 
  actionDispatcher, err := newActionDispatcher(managedApplication.bgContext, log, &handlerDefault{log: log})
@@ -183,13 +185,24 @@ func newManaged(
  },
  )
 
+ actionDispatcher.MustRegister(
+ &fleetapi.ActionUnenroll{},
+ &handlerUnenroll{
+ log: log,
+ emitter: emit,
+ dispatcher: router,
+ closers: []context.CancelFunc{managedApplication.cancelCtxFn},
+ },
+ )
+
  actionDispatcher.MustRegister(
  &fleetapi.ActionUnknown{},
  &handlerUnknown{log: log},
  )
 
  actions := actionStore.Actions()
- if len(actions) > 0 {
+
+ if len(actions) > 0 && !managedApplication.wasUnenrolled() {
  // TODO(ph) We will need an improvement on fleet, if there is an error while dispatching a
  // persisted action on disk we should be able to ask Fleet to get the latest configuration.
  // But at the moment this is not possible because the policy change was acked.
@@ -219,6 +232,11 @@ func newManaged(
 // Start starts a managed elastic-agent.
 func (m *Managed) Start() error {
  m.log.Info("Agent is starting")
+ if m.wasUnenrolled() {
+ m.log.Warnf("agent was previously unenrolled. To reactivate please reconfigure or enroll again.")
+ return nil
+ }
+
  m.gateway.Start()
  return nil
 }
@@ -236,3 +254,14 @@ func (m *Managed) Stop() error {
 func (m *Managed) AgentInfo() *info.AgentInfo {
  return m.agentInfo
 }
+
+func (m *Managed) wasUnenrolled() bool {
+ actions := m.as.Actions()
+ for _, a := range actions {
+ if a.Type() == "UNENROLL" {
+ return true
+ }
+ }
+
+ return false
+}

x-pack/elastic-agent/pkg/agent/application/managed_mode_test.go

@@ -12,6 +12,7 @@ import (
  "github.com/stretchr/testify/assert"
 
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/filters"
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/configrequest"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/fleetapi"
 )
@@ -68,16 +69,16 @@ func testActions() ([]action, error) {
 }
 
 type mockStreamStore struct {
- store []*configRequest
+ store []configrequest.Request
 }
 
 func newMockStreamStore() *mockStreamStore {
  return &mockStreamStore{
- store: make([]*configRequest, 0),
+ store: make([]configrequest.Request, 0),
  }
 }
 
-func (m *mockStreamStore) Execute(cr *configRequest) error {
+func (m *mockStreamStore) Execute(cr configrequest.Request) error {
  m.store = append(m.store, cr)
  return nil
 }

x-pack/elastic-agent/pkg/agent/application/router.go

@@ -7,7 +7,9 @@ package application
 import (
  "fmt"
  "strings"
+ "time"
 
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/configrequest"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/program"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/sorted"
@@ -19,7 +21,7 @@ var defautlRK = "DEFAULT"
 type routingKey = string
 
 type stream interface {
- Execute(*configRequest) error
+ Execute(configrequest.Request) error
  Close() error
  Shutdown()
 }
@@ -73,10 +75,7 @@ func (r *router) Dispatch(id string, grpProg map[routingKey][]program.Program) e
  return fmt.Errorf("could not find programs for routing key %s", rk)
  }
 
- req := &configRequest{
- id: id,
- programs: programs.([]program.Program),
- }
+ req := configrequest.New(id, time.Now(), programs.([]program.Program))
 
  r.log.Debugf(
  "Streams %s need to run config with ID %s and programs: %s",

x-pack/elastic-agent/pkg/agent/application/router_test.go

@@ -9,6 +9,7 @@ import (
 
  "github.com/stretchr/testify/require"
 
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/configrequest"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/program"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
 )
@@ -199,7 +200,7 @@ func newMockStream(rk routingKey, notify notifyFunc) *mockStream {
  }
 }
 
-func (m *mockStream) Execute(req *configRequest) error {
+func (m *mockStream) Execute(req configrequest.Request) error {
  m.event(executeOp, req)
  return nil
 }

x-pack/elastic-agent/pkg/agent/application/stream.go

@@ -7,6 +7,7 @@ package application
 import (
  "context"
 
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/configrequest"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/operation"
  operatorCfg "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/operation/config"
@@ -28,10 +29,10 @@ type operatorStream struct {
 }
 
 func (b *operatorStream) Close() error {
- return b.configHandler.HandleConfig(&configRequest{})
+ return b.configHandler.Close()
 }
 
-func (b *operatorStream) Execute(cfg *configRequest) error {
+func (b *operatorStream) Execute(cfg configrequest.Request) error {
  return b.configHandler.HandleConfig(cfg)
 }
 

x-pack/elastic-agent/pkg/agent/application/config_request.go → x-pack/elastic-agent/pkg/agent/configrequest/config_request.go

@@ -2,7 +2,7 @@
 // or more contributor license agreements. Licensed under the Elastic License;
 // you may not use this file except in compliance with the Elastic License.
 
-package application
+package configrequest
 
 import (
  "strings"
@@ -19,6 +19,15 @@ type configRequest struct {
  programs []program.Program
 }
 
+// New created a new Request.
+func New(id string, createdAt time.Time, programs []program.Program) Request {
+ return &configRequest{
+ id: id,
+ createdAt: createdAt,
+ programs: programs,
+ }
+}
+
 func (c *configRequest) String() string {
  names := c.ProgramNames()
  return "[" + c.ShortID() + "] Config: " + strings.Join(names, ", ")

x-pack/elastic-agent/pkg/agent/application/config_request_test.go → x-pack/elastic-agent/pkg/agent/configrequest/config_request_test.go

@@ -2,7 +2,7 @@
 // or more contributor license agreements. Licensed under the Elastic License;
 // you may not use this file except in compliance with the Elastic License.
 
-package application
+package configrequest
 
 import (
  "testing"

x-pack/elastic-agent/pkg/agent/configrequest/request.go

@@ -13,6 +13,8 @@ import (
 // Request is the minimal interface a config request must have.
 type Request interface {
  ID() string
+ ShortID() string
  CreatedAt() time.Time
  Programs() []program.Program
+ ProgramNames() []string
 }

x-pack/elastic-agent/pkg/agent/operation/monitoring.go

@@ -65,7 +65,7 @@ func (o *Operator) handleStartSidecar(s configrequest.Step) (result error) {
 }
 
 func (o *Operator) handleStopSidecar(s configrequest.Step) (result error) {
- for _, step := range o.getMonitoringSteps(s) {
+ for _, step := range o.generateMonitoringSteps(s.Version, nil) {
  p, _, err := getProgramFromStepWithTags(step, o.config.DownloadConfig, monitoringTags())
  if err != nil {
  return errors.New(err,

x-pack/elastic-agent/pkg/agent/operation/monitoring_test.go

@@ -175,6 +175,9 @@ func (b *testMonitor) EnrichArgs(_ string, _ string, args []string, _ bool) []st
 // Cleanup cleans up all drops.
 func (b *testMonitor) Cleanup(string, string) error { return nil }
 
+// Close closes the monitor.
+func (b *testMonitor) Close() {}
+
 // Prepare executes steps in order for monitoring to work correctly
 func (b *testMonitor) Prepare(string, string, int, int) error { return nil }