Skip to content

chore: separate actions from the controller service #203

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
arturaslcast merged 3 commits into from
Dec 29, 2025
Merged

chore: separate actions from the controller service #203

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
afc33f2
chore: separate actions from the controller service
arturaslcast Dec 28, 2025
6ebc0d3
rename some vars
arturaslcast Dec 29, 2025
3f6bf93
fix lint
arturaslcast Dec 29, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 11 additions & 3 deletions cmd/controller/run.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (

"github.com/castai/cluster-controller/cmd/utils"
"github.com/castai/cluster-controller/health"
"github.com/castai/cluster-controller/internal/actions"
"github.com/castai/cluster-controller/internal/actions/csr"
"github.com/castai/cluster-controller/internal/castai"
"github.com/castai/cluster-controller/internal/config"
Expand Down Expand Up @@ -131,6 +132,15 @@ func runController(

log.Infof("running castai-cluster-controller version %v, log-level: %v", binVersion, logger.Level)

actionHandlers := actions.NewDefaultActionHandlers(
k8sVer.Full(),
cfg.SelfPod.Namespace,
log,
clientset,
dynamicClient,
helmClient,
)

actionsConfig := controller.Config{
PollWaitInterval: 5 * time.Second,
PollTimeout: maxRequestTimeout,
Expand All @@ -148,11 +158,9 @@ func runController(
log,
actionsConfig,
k8sVer.Full(),
clientset,
dynamicClient,
client,
helmClient,
healthzAction,
actionHandlers,
)
defer func() {
if err := svc.Close(); err != nil {
Expand Down
44 changes: 44 additions & 0 deletions internal/actions/actions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
package actions

import (
"reflect"

"github.com/sirupsen/logrus"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"

"github.com/castai/cluster-controller/internal/castai"
"github.com/castai/cluster-controller/internal/helm"
)

type ActionHandlers map[reflect.Type]ActionHandler

func NewDefaultActionHandlers(
k8sVersion string,
castNamespace string,
log logrus.FieldLogger,
clientset *kubernetes.Clientset,
dynamicClient dynamic.Interface,
helmClient helm.Client,
) ActionHandlers {
return ActionHandlers{
reflect.TypeOf(&castai.ActionDeleteNode{}): NewDeleteNodeHandler(log, clientset),
reflect.TypeOf(&castai.ActionDrainNode{}): NewDrainNodeHandler(log, clientset, castNamespace),
reflect.TypeOf(&castai.ActionPatchNode{}): NewPatchNodeHandler(log, clientset),
reflect.TypeOf(&castai.ActionCreateEvent{}): NewCreateEventHandler(log, clientset),
reflect.TypeOf(&castai.ActionChartUpsert{}): NewChartUpsertHandler(log, helmClient),
reflect.TypeOf(&castai.ActionChartUninstall{}): NewChartUninstallHandler(log, helmClient),
reflect.TypeOf(&castai.ActionChartRollback{}): NewChartRollbackHandler(log, helmClient, k8sVersion),
reflect.TypeOf(&castai.ActionDisconnectCluster{}): NewDisconnectClusterHandler(log, clientset),
reflect.TypeOf(&castai.ActionCheckNodeDeleted{}): NewCheckNodeDeletedHandler(log, clientset),
reflect.TypeOf(&castai.ActionCheckNodeStatus{}): NewCheckNodeStatusHandler(log, clientset),
reflect.TypeOf(&castai.ActionEvictPod{}): NewEvictPodHandler(log, clientset),
reflect.TypeOf(&castai.ActionPatch{}): NewPatchHandler(log, dynamicClient),
reflect.TypeOf(&castai.ActionCreate{}): NewCreateHandler(log, dynamicClient),
reflect.TypeOf(&castai.ActionDelete{}): NewDeleteHandler(log, dynamicClient),
}
}

func (h ActionHandlers) Close() error {
return h[reflect.TypeOf(&castai.ActionCreateEvent{})].(*CreateEventHandler).Close()
Copy link
Copy Markdown
Contributor

@Tsonov Tsonov Dec 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this can panic, right?

Copy link
Copy Markdown
Contributor Author

@arturaslcast arturaslcast Dec 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not really when used with the constructor, but this is mostly copy & paste from the existing place

Copy link
Copy Markdown
Contributor

@Tsonov Tsonov Dec 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, maybe we should make it not panic even in this case, minor change

}
30 changes: 5 additions & 25 deletions internal/controller/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,10 @@ import (
"time"

"github.com/sirupsen/logrus"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"

"github.com/castai/cluster-controller/health"
"github.com/castai/cluster-controller/internal/actions"
"github.com/castai/cluster-controller/internal/castai"
"github.com/castai/cluster-controller/internal/helm"
"github.com/castai/cluster-controller/internal/metrics"
"github.com/castai/cluster-controller/internal/waitext"
)
Expand All @@ -38,35 +35,18 @@ func NewService(
log logrus.FieldLogger,
cfg Config,
k8sVersion string,
clientset *kubernetes.Clientset,
dynamicClient dynamic.Interface,
castaiClient castai.CastAIClient,
helmClient helm.Client,
healthCheck *health.HealthzProvider,
actionHandlers actions.ActionHandlers,
) *Controller {
return &Controller{
log: log,
cfg: cfg,
k8sVersion: k8sVersion,
castAIClient: castaiClient,
startedActions: map[string]struct{}{},
actionHandlers: map[reflect.Type]actions.ActionHandler{
reflect.TypeOf(&castai.ActionDeleteNode{}): actions.NewDeleteNodeHandler(log, clientset),
reflect.TypeOf(&castai.ActionDrainNode{}): actions.NewDrainNodeHandler(log, clientset, cfg.Namespace),
reflect.TypeOf(&castai.ActionPatchNode{}): actions.NewPatchNodeHandler(log, clientset),
reflect.TypeOf(&castai.ActionCreateEvent{}): actions.NewCreateEventHandler(log, clientset),
reflect.TypeOf(&castai.ActionChartUpsert{}): actions.NewChartUpsertHandler(log, helmClient),
reflect.TypeOf(&castai.ActionChartUninstall{}): actions.NewChartUninstallHandler(log, helmClient),
reflect.TypeOf(&castai.ActionChartRollback{}): actions.NewChartRollbackHandler(log, helmClient, cfg.Version),
reflect.TypeOf(&castai.ActionDisconnectCluster{}): actions.NewDisconnectClusterHandler(log, clientset),
reflect.TypeOf(&castai.ActionCheckNodeDeleted{}): actions.NewCheckNodeDeletedHandler(log, clientset),
reflect.TypeOf(&castai.ActionCheckNodeStatus{}): actions.NewCheckNodeStatusHandler(log, clientset),
reflect.TypeOf(&castai.ActionEvictPod{}): actions.NewEvictPodHandler(log, clientset),
reflect.TypeOf(&castai.ActionPatch{}): actions.NewPatchHandler(log, dynamicClient),
reflect.TypeOf(&castai.ActionCreate{}): actions.NewCreateHandler(log, dynamicClient),
reflect.TypeOf(&castai.ActionDelete{}): actions.NewDeleteHandler(log, dynamicClient),
},
healthCheck: healthCheck,
actionHandlers: actionHandlers,
healthCheck: healthCheck,
}
}

Expand All @@ -77,7 +57,7 @@ type Controller struct {

k8sVersion string

actionHandlers map[reflect.Type]actions.ActionHandler
actionHandlers actions.ActionHandlers

startedActionsWg sync.WaitGroup
startedActions map[string]struct{}
Expand Down Expand Up @@ -273,5 +253,5 @@ func getHandlerError(err error) *string {
}

func (s *Controller) Close() error {
return s.actionHandlers[reflect.TypeOf(&castai.ActionCreateEvent{})].(*actions.CreateEventHandler).Close()
return s.actionHandlers.Close()
}
36 changes: 20 additions & 16 deletions internal/controller/controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package controller
import (
"context"
"fmt"
"reflect"
"testing"
"time"

Expand All @@ -11,12 +12,12 @@ import (
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/require"
"go.uber.org/goleak"
"k8s.io/client-go/kubernetes"

"github.com/castai/cluster-controller/health"
"github.com/castai/cluster-controller/internal/actions"
mock_actions "github.com/castai/cluster-controller/internal/actions/mock"
"github.com/castai/cluster-controller/internal/castai"
"github.com/castai/cluster-controller/internal/castai/mock"
mock_castai "github.com/castai/cluster-controller/internal/castai/mock"
)

// nolint: govet
Expand Down Expand Up @@ -99,9 +100,9 @@ func TestController_Run(t *testing.T) {
},
},
}, nil).Times(1).MinTimes(1)
m.EXPECT().AckAction(gomock.Any(), "a1", gomock.Any()).Return(nil).MinTimes(1)
m.EXPECT().AckAction(gomock.Any(), "a2", gomock.Any()).Return(nil).MinTimes(1)
m.EXPECT().AckAction(gomock.Any(), "a3", gomock.Any()).Return(nil).MinTimes(1)
m.EXPECT().AckAction(gomock.Any(), "a1", &castai.AckClusterActionRequest{}).Return(nil).MinTimes(1)
m.EXPECT().AckAction(gomock.Any(), "a2", &castai.AckClusterActionRequest{}).Return(nil).MinTimes(1)
m.EXPECT().AckAction(gomock.Any(), "a3", &castai.AckClusterActionRequest{}).Return(nil).MinTimes(1)
},
},
},
Expand Down Expand Up @@ -240,22 +241,25 @@ func TestController_Run(t *testing.T) {
if tt.fields.tuneMockCastAIClient != nil {
tt.fields.tuneMockCastAIClient(client)
}
s := NewService(
logrus.New(),
tt.fields.cfg,
tt.fields.k8sVersion,
kubernetes.New(nil),
nil,
client,
nil,
health.NewHealthzProvider(health.HealthzCfg{HealthyPollIntervalLimit: pollTimeout}, logrus.New()))

handler := mock_actions.NewMockActionHandler(m)
if tt.fields.tuneMockHandler != nil {
tt.fields.tuneMockHandler(handler)
}
for k := range s.actionHandlers {
s.actionHandlers[k] = handler
testActionHandlers := map[reflect.Type]actions.ActionHandler{
reflect.TypeOf(&castai.ActionDeleteNode{}): handler,
Copy link
Copy Markdown
Contributor

@Tsonov Tsonov Dec 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We used to initialize them all with mock handler, correct? Does this mean we only test these 3?

Copy link
Copy Markdown
Contributor Author

@arturaslcast arturaslcast Dec 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

correct, only 3 action types are actually tested here. I'd argue if only fake actions should be used instead as it's mostly to test if the controller implementation works correctly

reflect.TypeOf(&castai.ActionDrainNode{}): handler,
reflect.TypeOf(&castai.ActionPatchNode{}): handler,
}

s := NewService(
logrus.New(),
tt.fields.cfg,
tt.fields.k8sVersion,
client,
health.NewHealthzProvider(health.HealthzCfg{HealthyPollIntervalLimit: pollTimeout}, logrus.New()),
testActionHandlers)

s.Run(tt.args.ctx())
})
}
Expand Down
Loading