O framework controller-runtime é uma abstração robusta fornecida pela comunidade Kubernetes para simplificar o desenvolvimento de controladores. Embora sua lógica fundamental se assemelhe à de um controlador personalizado, ele introduz conceitos e utilitários que agilizam o processo para os desenvolvedores. Ferramentas populares como kubebuilder e operator-sdk, por exemplo, são construídas sobre o controller-runtime, oferecendo andaimes para projetos e acelerando o desenvolvimento.

Este artigo explora a implementação interna de um controlador dentro do controller-runtime, detalhando como ele gerencia o ciclo de vida e a reconciliação de recursos.

Estrutura do Controlador

Vamos primeiramente analisar a estrutura fundamental de um controlador. A definição da struct principal de um controlador é a seguinte:

// pkg/internal/controller/controller.go

// Controller implementa a interface controller.Controller.
type Controller struct {
    // Name é um identificador único para o Controlador, usado em rastreamento, logs e monitoramento. É um campo obrigatório.
    Name string

    // MaxConcurrentReconciles define o número máximo de operações de reconciliação que podem ser executadas
    // concomitantemente. O valor padrão é 1.
    MaxConcurrentReconciles int

    // Do é a função de reconciliação, que pode ser invocada a qualquer momento com o nome e namespace de um objeto.
    // Ela assegura que o estado do sistema corresponda ao estado desejado especificado no objeto.
    // Por padrão, utiliza DefaultReconcileFunc.
    Do reconcile.Reconciler

    // MakeQueue constrói a fila de trabalho para este controlador assim que ele estiver pronto para iniciar.
    // Esta abordagem é adotada porque as workqueues padrão do Kubernetes iniciam-se imediatamente,
    // o que poderia levar a vazamento de goroutines se controller.New fosse chamado repetidamente.
    MakeQueue func() workqueue.RateLimitingInterface

    // Queue é uma listeningQueue que escuta eventos dos Informers e adiciona chaves de objeto à fila para processamento.
    // A propriedade MakeQueue é responsável por construir esta fila de trabalho.
    Queue workqueue.RateLimitingInterface

    // SetFields é usado para injetar dependências em outros objetos, como Sources, EventHandlers e Predicates.
    // (Obsoleto: o chamador deve lidar com a injeção de campos por conta própria.)
    SetFields func(i interface{}) error

    // mu é um mutex para sincronizar a configuração do Controlador.
    mu sync.Mutex

    // Started indica se o Controlador foi iniciado.
    Started bool

    // ctx é o contexto passado para Start() e usado ao iniciar os watches.
    // Embora o armazenamento de contextos em structs seja geralmente desaconselhado, esta é considerada uma situação legada.
    ctx context.Context

    // CacheSyncTimeout define o tempo limite para esperar pela sincronização do cache.
    // Se não for configurado, o padrão é de 2 minutos.
    CacheSyncTimeout time.Duration

    // startWatches mantém uma lista de fontes (sources), manipuladores (handlers) e predicados (predicates)
    // a serem iniciados quando o controlador for ativado.
    startWatches []watchDescription

    // LogConstructor é utilizado para criar um logger para registrar mensagens durante a reconciliação,
    // ou ao iniciar um watch.
    // Observação: LogConstructor deve ser capaz de lidar com requisições nulas, pois também é usado
    // fora do contexto de uma reconciliação.
    LogConstructor func(request *reconcile.Request) logr.Logger

    // RecoverPanic indica se panics causados pela reconciliação devem ser recuperados.
    RecoverPanic bool
}

A struct Controller acima encapsula a lógica central de um controlador no controller-runtime. Observamos a presença de uma fila de trabalho com limitação de taxa (workqueue.RateLimitingInterface), crucial para o processamento de eventos. No entanto, ela não expõe diretamente um Informer ou Indexer para recursos. Em vez disso, essa funcionalidade é abstraída através do campo startWatches, que é uma fatia de structs watchDescription. Cada watchDescription contém as informações essenciais necessárias para configurar um watch:

// pkg/internal/controller/controller.go

// watchDescription contém todas as informações necessárias para iniciar um watch.
type watchDescription struct {
    src        source.Source
    handler    handler.EventHandler
    predicates []predicate.Predicate
}

Dentro do ciclo de vida de um controlador, as funções mais críticas são Watch e Start. A seguir, detalharemos a implementação de cada uma delas.

Implementação da Função Watch

A função Watch é responsável por registrar as fontes de eventos e os manipuladores correspondentes. Sua implementação no controller-runtime é apresentada abaixo:

// pkg/internal/controller/controller.go

// Watch implementa controller.Controller.
func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prct ...predicate.Predicate) error {
    c.mu.Lock()
    defer c.mu.Unlock()

    // Injeta o Cache nos argumentos, se necessário
    if err := c.SetFields(src); err != nil {
        return err
    }
    if err := c.SetFields(evthdler); err != nil {
        return err
    }
    for _, pr := range prct {
        if err := c.SetFields(pr); err != nil {
            return err
        }
    }

    // Se o Controlador ainda não foi iniciado, armazena os watches localmente e retorna.
    // Estes watches serão mantidos na struct do controlador até que o manager ou o usuário invoque Start(...).
    if !c.Started {
        c.startWatches = append(c.startWatches, watchDescription{src: src, handler: evthdler, predicates: prct})
        return nil
    }

    c.LogConstructor(nil).Info("Iniciando EventSource", "source", src)
    // Invoca a função Start da fonte (src).
    return src.Start(c.ctx, evthdler, c.Queue, prct...)
}

Como observado na função Watch, o processo culmina na invocação do método Start do parâmetro source.Source. Este Source atua como a origem dos eventos, transformando operações como Create, Update ou Delete em recursos em reconcile.Requests que são enfileirados por event.EventHandlers.

O controller-runtime suporta diferentes tipos de fontes de eventos:

• Kind: Para eventos originados de recursos dentro do cluster (ex: criação de Pod, atualização de Deployment).
• Channel: Para eventos provenientes de fontes externas ao cluster (ex: callbacks de webhooks do GitHub, sondagem de URLs externas).

A interface source.Source é definida como:

// pkg/source/source.go

type Source interface {
	// Start é uma função interna e deve ser invocada apenas pelo Controller para registrar um EventHandler
	// com o Informer, a fim de enfileirar reconcile.Requests.
	Start(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
}

Para entender a implementação específica do método Start da interface Source, precisamos examinar a implementação concreta que é passada para Controller.Watch. Em um cenário comum com controller-runtime, a chamada para Controller.Watch é orquestrada pela função doWatch(), localizada em pkg/builder/controller.go:

// pkg/builder/controller.go

func (blder *Builder) doWatch() error {
	// Tipo de recurso para reconciliação
	src := &source.Kind{Type: blder.forInput.object}
	hdler := &handler.EnqueueRequestForObject{}
	allPredicates := append(blder.globalPredicates, blder.forInput.predicates...)
	err := blder.ctrl.Watch(src, hdler, allPredicates...)
	if err != nil {
		return err
	}
  // ... outras lógicas ...
	return nil
}

Aqui, o primeiro argumento para Watch é uma instância de source.Kind. Esta struct implementa a interface source.Source:

// pkg/source/source.go

// Kind fornece uma fonte de eventos originados no cluster a partir de Watches (ex: Criação de Pod).
type Kind struct {
    // Type é o tipo de objeto a ser monitorado. Ex: &v1.Pod{}
    Type client.Object

    // cache é usado para monitorar APIs.
    cache cache.Cache

    // started pode conter um erro se um foi encontrado durante a inicialização. Se fechado e sem erro,
    // a inicialização e sincronização foram concluídas.
    started     chan error
    startCancel func()
}

// Start é uma função interna e deve ser invocada apenas pelo Controller para registrar um EventHandler
// com o Informer, a fim de enfileirar reconcile.Requests.
func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface,
    prct ...predicate.Predicate) error {
    // O tipo (Type) deve ter sido especificado pelo usuário.
    if ks.Type == nil {
        return fmt.Errorf("deve especificar Kind.Type")
    }

    // O cache deve ter sido injetado antes da chamada de Start.
    if ks.cache == nil {
        return fmt.Errorf("deve chamar CacheInto em Kind antes de chamar Start")
    }

    // cache.GetInformer bloqueará até que seu contexto seja cancelado se o cache já estiver iniciado e
    // não conseguir sincronizar esse informer (geralmente devido a problemas de RBAC).
    ctx, ks.startCancel = context.WithCancel(ctx)
    ks.started = make(chan error)
    go func() {
        var (
            i       cache.Informer
            lastErr error
        )

        // Tenta obter um informer até que ele retorne true, um erro ou o contexto especificado seja cancelado ou expire.
        if err := wait.PollImmediateUntilWithContext(ctx, 10*time.Second, func(ctx context.Context) (bool, error) {
            // Busca o Informer no Cache e adiciona um EventHandler que preenche a fila.
            i, lastErr = ks.cache.GetInformer(ctx, ks.Type)
            if lastErr != nil {
                kindMatchErr := &meta.NoKindMatchError{}
                switch {
                case errors.As(lastErr, &kindMatchErr):
                    log.Error(lastErr, "se o tipo é um CRD, ele deve ser instalado antes de chamar Start",
                        "kind", kindMatchErr.GroupKind)
                case runtime.IsNotRegisteredError(lastErr):
                    log.Error(lastErr, "o tipo deve ser registrado no Scheme")
                default:
                    log.Error(lastErr, "falha ao obter informer do cache")
                }
                return false, nil // Tentar novamente.
            }
            return true, nil
        }); err != nil {
            if lastErr != nil {
                ks.started <- fmt.Errorf("falha ao obter informer do cache: %w", lastErr)
                return
            }
            ks.started <- err
            return
        }

        i.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
        if !ks.cache.WaitForCacheSync(ctx) {
            // Seria ótimo retornar algo mais informativo aqui
            ks.started <- errors.New("cache não sincronizou")
        }
        close(ks.started)
    }()

    return nil
}

A partir desta implementação detalhada, fica claro que a função Controller.Watch é essencialmente responsável por adquirir o Informer para o tipo de recurso desejado e registrar as funções de escuta de eventos. O Informer é obtido através de uma instância de cache, que é injetada antes da chamada de Start. Uma vez obtido o Informer, AddEventHandler é invocado com uma struct internal.EventHandler. Esta struct, por sua vez, implementa a interface ResourceEventHandler do client-go, expondo os métodos familiares OnAdd, OnUpdate e OnDelete:

// pkg/source/internal/eventsource.go

// EventHandler adapta a interface handler.EventHandler para a interface cache.ResourceEventHandler.
type EventHandler struct {
    EventHandler handler.EventHandler
    Queue        workqueue.RateLimitingInterface
    Predicates   []predicate.Predicate
}

// OnAdd cria um CreateEvent e invoca o método Create no EventHandler.
func (e EventHandler) OnAdd(obj interface{}) {
    // Evento de criação de objeto Kubernetes.
    c := event.CreateEvent{}

    // Extrai o Object do item.
    if o, ok := obj.(client.Object); ok {
        c.Object = o
    } else {
        log.Error(nil, "OnAdd: Objeto ausente no CreateEvent",
            "object", obj, "type", fmt.Sprintf("%T", obj))
        return
    }

    // Predicates são usados para filtragem de eventos, invocando a função Create de cada Predicate.
    for _, p := range e.Predicates {
        if !p.Create(c) {
            return
        }
    }

    // Invoca o manipulador de criação.
    e.EventHandler.Create(c, e.Queue)
}

// OnUpdate cria um UpdateEvent e invoca o método Update no EventHandler.
func (e EventHandler) OnUpdate(oldObj, newObj interface{}) {
    // Evento de atualização.
    u := event.UpdateEvent{}

    if o, ok := oldObj.(client.Object); ok {
        u.ObjectOld = o
    } else {
        log.Error(nil, "OnUpdate: Objeto Antigo ausente no UpdateEvent",
            "object", oldObj, "type", fmt.Sprintf("%T", oldObj))
        return
    }

    // Extrai o novo Objeto do item.
    if o, ok := newObj.(client.Object); ok {
        u.ObjectNew = o
    } else {
        log.Error(nil, "OnUpdate: Objeto Novo ausente no UpdateEvent",
            "object", newObj, "type", fmt.Sprintf("%T", newObj))
        return
    }

    for _, p := range e.Predicates {
        if !p.Update(u) {
            return
        }
    }

    // Invoca o manipulador de atualização.
    e.EventHandler.Update(u, e.Queue)
}

// OnDelete cria um DeleteEvent e invoca o método Delete no EventHandler.
func (e EventHandler) OnDelete(obj interface{}) {
    d := event.DeleteEvent{}

    // Lida com eventos de "tombstone" (objetos deletados com estado final desconhecido) extraindo o objeto.
    // Eventos de tombstone encapsulam o objeto em uma struct DeleteFinalStateUnknown, então o objeto precisa ser extraído.
    // Copiado de sample-controller.
    var ok bool
    if _, ok = obj.(client.Object); !ok {
        // Se o objeto não tem Metadata, assume-se que é um objeto tombstone do tipo DeletedFinalStateUnknown.
        tombstone, tombstoneOk := obj.(cache.DeletedFinalStateUnknown)
        if !tombstoneOk {
            log.Error(nil, "Erro ao decodificar objetos. Esperado cache.DeletedFinalStateUnknown",
                "type", fmt.Sprintf("%T", obj),
                "object", obj)
            return
        }

        // Define obj para o objeto tombstone.
        obj = tombstone.Obj
    }

    // Extrai o Object do item.
    if o, ok := obj.(client.Object); ok {
        d.Object = o
    } else {
        log.Error(nil, "OnDelete: Objeto ausente no DeleteEvent",
            "object", obj, "type", fmt.Sprintf("%T", obj))
        return
    }

    for _, p := range e.Predicates {
        if !p.Delete(d) {
            return
        }
    }

    // Invoca o manipulador de exclusão.
    e.EventHandler.Delete(d, e.Queue)
}

Dentro dos métodos OnAdd, OnUpdate e OnDelete do internal.EventHandler, a primeira etapa crucial é a filtragem de eventos pelos Predicates. Somente após a aprovação pelos predicados, o evento é processado pelo handler.EventHandler. Este manipulador é, conforme visto na função doWatch(), uma instância de &handler.EnqueueRequestForObject{}. Assim, a lógica de enfileiramento reside nos métodos desta struct:

// pkg/handler/enqueue.go

// EnqueueRequestForObject enfileira um Request contendo o Nome e Namespace do objeto que é a fonte do Evento.
// (ex: Nome e Namespace dos objetos criados / deletados / atualizados).
// handler.EnqueueRequestForObject é utilizado por quase todos os Controladores que possuem Recursos associados (ex: CRDs)
// para reconciliar o Recurso associado.
type EnqueueRequestForObject struct{}


// Create implementa EventHandler.
func (e *EnqueueRequestForObject) Create(evt event.CreateEvent, q workqueue.RateLimitingInterface) {
    if evt.Object == nil {
        enqueueLog.Error(nil, "CreateEvent recebido sem metadados", "event", evt)
        return
    }
    // Adiciona um objeto Request à fila de trabalho.
    q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
        Name:      evt.Object.GetName(),
        Namespace: evt.Object.GetNamespace(),
    }})
}

// Update implementa EventHandler.
func (e *EnqueueRequestForObject) Update(evt event.UpdateEvent, q workqueue.RateLimitingInterface) {
    switch {
    // Se o novo objeto não for nulo, adiciona-o à fila de trabalho.
    case evt.ObjectNew != nil:
        q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
            Name:      evt.ObjectNew.GetName(),
            Namespace: evt.ObjectNew.GetNamespace(),
        }})
    // Se o objeto antigo existir, adiciona-o à fila de trabalho.
    case evt.ObjectOld != nil:
        q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
            Name:      evt.ObjectOld.GetName(),
            Namespace: evt.ObjectOld.GetNamespace(),
        }})
    default:
        enqueueLog.Error(nil, "UpdateEvent recebido sem metadados", "event", evt)
    }
}

// Delete implementa EventHandler.
func (e *EnqueueRequestForObject) Delete(evt event.DeleteEvent, q workqueue.RateLimitingInterface) {
    if evt.Object == nil {
        enqueueLog.Error(nil, "DeleteEvent recebido sem metadados", "event", evt)
        return
    }
    // Como o estado de exclusão do objeto já foi tratado, ele é diretamente colocado na fila.
    q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
        Name:      evt.Object.GetName(),
        Namespace: evt.Object.GetNamespace(),
    }})
}

Os métodos Create, Update e Delete da struct EnqueueRequestForObject demonstram que os itens enfileirados não são simplesmente chaves únicas, mas sim objetos reconcile.Request encapsulados, que permitem o fácil acesso à identificação única do objeto (NamespacedName). Em resumo, a função Controller.Watch do controller-runtime abstrai e implementa o processo de inicialização de Informers e o registro de manipuladores de eventos, um conceito fundamental em qualquer controlador Kubernetes.

Implementação da Função Start

Após a configuração do mecanismo de observação (Watch), o próximo passo é a ativação do controlador, que é orquestrada pela função Controller.Start. Esta função é análoga ao loop de execução principal em controladores personalizados.

// pkg/internal/controller/controller.go

// Start implementa controller.Controller.
func (c *Controller) Start(ctx context.Context) error {
    c.mu.Lock()
    // Verifica se o controlador já foi iniciado; caso sim, retorna um erro para evitar múltiplas inicializações.
    if c.Started {
        return errors.New("o controlador foi iniciado mais de uma vez. Isso provavelmente é causado por ser adicionado a um manager múltiplas vezes")
    }

    c.initMetrics()

    // Configura o contexto interno.
    c.ctx = ctx

    // Invoca MakeQueue() para gerar a fila de trabalho.
    c.Queue = c.MakeQueue()
    go func() {
        <-ctx.Done()
        c.Queue.ShutDown()
    }()

    wg := &sync.WaitGroup{}
    err := func() error {
        defer c.mu.Unlock()

        // TODO(pwittrock): Reconsiderar HandleCrash
        defer utilruntime.HandleCrash()

        // NB(directxman12): Inicia as fontes *antes* de tentar esperar pela sincronização dos caches,
        // para que elas tenham a chance de registrar seus caches pretendidos.
        for _, watch := range c.startWatches {
            c.LogConstructor(nil).Info("Iniciando EventSource", "source", fmt.Sprintf("%s", watch.src))

            if err := watch.src.Start(ctx, watch.handler, c.Queue, watch.predicates...); err != nil {
                return err
            }
        }

        // Inicia as fábricas SharedIndexInformer para começar a popular os caches SharedIndexInformer.
        c.LogConstructor(nil).Info("Iniciando Controlador")

        for _, watch := range c.startWatches {
            syncingSource, ok := watch.src.(source.SyncingSource)
            if !ok {
                continue
            }

            if err := func() error {
                // Usa um contexto com timeout para iniciar as fontes e sincronizar os caches.
                sourceStartCtx, cancel := context.WithTimeout(ctx, c.CacheSyncTimeout)
                defer cancel()

                // WaitForSync aguarda um tempo limite definitivo e retorna se houver um erro ou timeout.
                // Aguarda a sincronização do Informer.
                if err := syncingSource.WaitForSync(sourceStartCtx); err != nil {
                    err := fmt.Errorf("falha ao aguardar a sincronização dos caches de %s: %w", c.Name, err)
                    c.LogConstructor(nil).Error(err, "Não foi possível aguardar a sincronização do Cache")
                    return err
                }

                return nil
            }(); err != nil {
                return err
            }
        }

        // Todos os watches foram iniciados; podemos redefinir a fatia local.
        // Não devemos manter watches mais do que o necessário, pois cada fonte de watch pode conter um cache subjacente
        // que não será coletado pelo garbage collector se mantivermos uma referência a ele.
        c.startWatches = nil

        // Lança workers para processar recursos.
        c.LogConstructor(nil).Info("Iniciando workers", "worker count", c.MaxConcurrentReconciles)
        wg.Add(c.MaxConcurrentReconciles)
        for i := 0; i < c.MaxConcurrentReconciles; i++ {
            go func() {
                defer wg.Done()
                // Executa um thread worker que simplesmente desenfileira itens, os processa e os marca como concluídos.
                // Garante que o reconcileHandler nunca seja invocado concomitantemente com o mesmo objeto.
                for c.processNextWorkItem(ctx) {
                }
            }()
        }

        c.Started = true
        return nil
    }()
    if err != nil {
        return err
    }

    <-ctx.Done()
    c.LogConstructor(nil).Info("Sinal de desligamento recebido, aguardando todos os workers terminarem")
    wg.Wait()
    c.LogConstructor(nil).Info("Todos os workers terminaram")
    return nil
}

A função Start, similar aos controladores customizados, primeiramente aguarda a sincronização completa dos Informers de recursos. Em seguida, ela inicia um conjunto de "workers" (goroutines) para processar os itens da fila de trabalho. O mecanismo de execução dos workers é idêntico ao padrão observado em controladores Kubernetes:

// pkg/internal/controller/controller.go

// processNextWorkItem lerá um único item da workqueue e tentará processá-lo,
// chamando o reconcileHandler.
func (c *Controller) processNextWorkItem(ctx context.Context) bool {
    // Retira um elemento da fila.
    obj, shutdown := c.Queue.Get()
    if shutdown {
        // Para o trabalho: a fila foi desligada.
        return false
    }

    // Chamamos Done aqui para que a workqueue saiba que terminamos de processar este item.
    // Também devemos lembrar de chamar Forget se não quisermos que este item de trabalho seja
    // reenfileirado. Por exemplo, não chamamos Forget se ocorrer um erro transitório;
    // em vez disso, o item é colocado de volta na workqueue e tentado novamente após um período de back-off.
    defer c.Queue.Done(obj)

    ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(1)
    defer ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(-1)

    // Invoca reconcileHandler para processar o elemento.
    c.reconcileHandler(ctx, obj)
    return true
}

// reconcileHandler processa um único item da fila de trabalho, orquestrando a lógica de reconciliação.
func (c *Controller) reconcileHandler(ctx context.Context, obj interface{}) {
    // Atualiza métricas após o processamento de cada item.
    reconcileStartTS := time.Now()
    defer func() {
        c.updateMetrics(time.Since(reconcileStartTS))
    }()

    // Garante que o objeto seja um request válido.
    req, ok := obj.(reconcile.Request)
    if !ok {
        // Se o item na workqueue for inválido, chamamos Forget aqui para evitar um loop de
        // tentativa de processar um item inválido.
        c.Queue.Forget(obj)
        c.LogConstructor(nil).Error(nil, "Item da fila não era um Request válido", "type", fmt.Sprintf("%T", obj), "value", obj)
        return
    }

    log := c.LogConstructor(&req)

    log = log.WithValues("reconcileID", uuid.NewUUID())
    ctx = logf.IntoContext(ctx, log)

    // Executa o syncHandler do RunInformersAndControllers, passando a string Namespace/Name do recurso a ser sincronizado.
    // Invoca a função Reconciler para processar este item, onde a lógica de negócio é implementada.
    result, err := c.Do.Reconcile(ctx, req)
    switch {
    case err != nil:
        // Se a lógica de negócio retornar um erro, o item é reenfileirado na fila com limitação de taxa.
        c.Queue.AddRateLimited(req)
        // Registra métricas de erro.
        ctrlmetrics.ReconcileErrors.WithLabelValues(c.Name).Inc()
        ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, labelError).Inc()
        log.Error(err, "Erro no Reconciler")

    // Se o resultado da função de reconciliação (Reconcile) incluir um RequestAfter maior que 0.
    case result.RequeueAfter > 0:
        // A requisição result.RequeueAfter será perdida se for retornada junto com um erro não-nil.
        // Mas isso é intencional, pois precisamos atingir loops de reconciliação estáveis antes de enfileirar
        // devido a result.RequestAfter.
        // "Esquece" o elemento.
        c.Queue.Forget(obj)
        // Adiciona à fila com um atraso.
        c.Queue.AddAfter(req, result.RequeueAfter)
        ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, labelRequeueAfter).Inc()
    case result.Requeue:
        // Adiciona à fila com limitação de taxa.
        c.Queue.AddRateLimited(req)
        ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, labelRequeue).Inc()
    default:
        // Finalmente, se nenhum erro ocorrer, "esquecemos" este item para que não seja
        // enfileirado novamente até que outra mudança aconteça.
        c.Queue.Forget(obj)
        ctrlmetrics.ReconcileTotal.WithLabelValues(c.Name, labelSuccess).Inc()
    }
}

A função reconcileHandler é o ponto central onde a lógica de negócio do controlador é executada. Ela engloba não apenas o processamento de eventos, mas também o tratamento de erros e a gestão do ciclo de vida da reconciliação. A parte essencial que o desenvolvedor implementa é o método c.Do.Reconcile(ctx, req). A decisão de reenfileirar ou descartar um item é baseada no valor de retorno desta função:

• Se um erro for retornado (err != nil), o item é reenfileirado na fila com limitação de taxa.
• Se result.RequeueAfter > 0, o item é "esquecido" pela fila e reenfileirado após o período especificado.
• Se result.Requeue for verdadeiro, o item é reenfileirado na fila com limitação de taxa.
• Se a reconciliação for concluída sem erros e sem requisições de reenfileiramento explícitas, o item é "esquecido" pela fila, aguardando um novo evento para ser processado novamente.