[Ingest Manager] Unroll on unauthorised

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

@@ -142,6 +142,7 @@ var _ actionConfigChangeSerializer = actionConfigChangeSerializer(fleetapi.Actio
 type actionUnenrollSerializer struct {
 	ActionID   string `yaml:"action_id"`
 	ActionType string `yaml:"action_type"`
+	IsDetected bool   `yaml:"is_detected"`
 }
 
 // Add a guards between the serializer structs and the original struct.

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

@@ -6,6 +6,7 @@ package application
 
 import (
 	"context"
+	"strings"
 	"sync"
 	"time"
 
@@ -17,6 +18,8 @@ import (
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/scheduler"
 )
 
+const maxUnauthCounter int = 6
+
 // Default Configuration for the Fleet Gateway.
 var defaultGatewaySettings = &fleetGatewaySettings{
 	Duration: 1 * time.Second,        // time between successful calls
@@ -60,18 +63,19 @@ type fleetAcker interface {
 // call the API to send the events and will receive actions to be executed locally.
 // The only supported action for now is a "ActionPolicyChange".
 type fleetGateway struct {
-	bgContext  context.Context
-	log        *logger.Logger
-	dispatcher dispatcher
-	client     clienter
-	scheduler  scheduler.Scheduler
-	backoff    backoff.Backoff
-	settings   *fleetGatewaySettings
-	agentInfo  agentInfo
-	reporter   fleetReporter
-	done       chan struct{}
-	wg         sync.WaitGroup
-	acker      fleetAcker
+	bgContext     context.Context
+	log           *logger.Logger
+	dispatcher    dispatcher
+	client        clienter
+	scheduler     scheduler.Scheduler
+	backoff       backoff.Backoff
+	settings      *fleetGatewaySettings
+	agentInfo     agentInfo
+	reporter      fleetReporter
+	done          chan struct{}
+	wg            sync.WaitGroup
+	acker         fleetAcker
+	unauthCounter int
 }
 
 func newFleetGateway(
@@ -210,6 +214,20 @@ func (f *fleetGateway) execute(ctx context.Context) (*fleetapi.CheckinResponse,
 	}
 
 	resp, err := cmd.Execute(ctx, req)
+	if isUnauth(err) {
+		f.unauthCounter++
+		if f.shouldUnroll() {
+			f.log.Warnf("retrieved unauthorized for '%d' times. Unrolling.", f.unauthCounter)
+			return &fleetapi.CheckinResponse{
+				Actions: []fleetapi.Action{&fleetapi.ActionUnenroll{ActionID: "", ActionType: "UNENROLL", IsDetected: true}},
+				Success: true,
+			}, nil
+		}
+
+		return nil, err
+	}
+
+	f.unauthCounter = 0
 	if err != nil {
 		return nil, err
 	}
@@ -219,6 +237,18 @@ func (f *fleetGateway) execute(ctx context.Context) (*fleetapi.CheckinResponse,
 	return resp, nil
 }
 
+func (f *fleetGateway) shouldUnroll() bool {
+	return f.unauthCounter >= maxUnauthCounter
+}
+
+func isUnauth(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	return strings.Contains(err.Error(), fleetapi.ErrInvalidAPIKey.Error())
+}
+
 func (f *fleetGateway) Start() {
 	f.wg.Add(1)
 	go func(wg *sync.WaitGroup) {

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

@@ -16,10 +16,11 @@ import (
 // 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
+	log         *logger.Logger
+	emitter     emitterFunc
+	dispatcher  programsDispatcher
+	closers     []context.CancelFunc
+	actionStore *actionStore
 }
 
 func (h *handlerUnenroll) Handle(ctx context.Context, a action, acker fleetAcker) error {
@@ -33,24 +34,26 @@ func (h *handlerUnenroll) Handle(ctx context.Context, a action, acker fleetAcker
 	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
+	if !action.IsDetected {
+		// ACK only events comming from fleet
+		if err := acker.Ack(ctx, action); err != nil {
+			return err
+		}
+
+		// commit all acks before quitting.
+		if err := acker.Commit(ctx); err != nil {
+			return err
+		}
+	} else if h.actionStore != nil {
+		// backup action for future start to avoid starting fleet gateway loop
+		h.actionStore.Add(a)
+		h.actionStore.Save()
 	}
 
 	// 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/managed_mode.go

@@ -183,10 +183,11 @@ func newManaged(
 	actionDispatcher.MustRegister(
 		&fleetapi.ActionUnenroll{},
 		&handlerUnenroll{
-			log:        log,
-			emitter:    emit,
-			dispatcher: router,
-			closers:    []context.CancelFunc{managedApplication.cancelCtxFn},
+			log:         log,
+			emitter:     emit,
+			dispatcher:  router,
+			closers:     []context.CancelFunc{managedApplication.cancelCtxFn},
+			actionStore: actionStore,
 		},
 	)
 

x-pack/elastic-agent/pkg/fleetapi/action.go

@@ -87,6 +87,7 @@ func (a *ActionConfigChange) ID() string {
 type ActionUnenroll struct {
 	ActionID   string
 	ActionType string
+	IsDetected bool
 }
 
 func (a *ActionUnenroll) String() string {