Skip to content

Commit

Permalink
Call reconcile from server (#58)
Browse files Browse the repository at this point in the history
* calling reconcile from server

* reconcile if crd is not found in CRDDirectory

* remove ForceReconcile function

* fix goimports lint and update error messages

* undo accidental change

* fix goimports lint

---------

Co-authored-by: Shubham Rai <[email protected]>
  • Loading branch information
Maanas-23 and shubhamrai1993 authored Jan 2, 2025
1 parent 3c2e568 commit 77fa407
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 7 deletions.
7 changes: 4 additions & 3 deletions operator/cmd/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -175,19 +175,20 @@ func mainWithError() error {
informerManager.Start()

// Set up the ElastiService controller
if err = (&controller.ElastiServiceReconciler{
reconciler := &controller.ElastiServiceReconciler{
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
Logger: zapLogger,
InformerManager: informerManager,
}).SetupWithManager(mgr); err != nil {
}
if err = reconciler.SetupWithManager(mgr); err != nil {
setupLog.Error(err, "unable to create controller", "controller", "ElastiService")
sentry.CaptureException(err)
return fmt.Errorf("main: %w", err)
}

// Start the elasti server
eServer := elastiserver.NewServer(zapLogger, mgr.GetConfig(), 30*time.Second)
eServer := elastiserver.NewServer(zapLogger, mgr.GetConfig(), 30*time.Second, reconciler)
errChan := make(chan error, 1)
go func() {
if err := eServer.Start(elastiServerPort); err != nil {
Expand Down
29 changes: 25 additions & 4 deletions operator/internal/elastiserver/elastiServer.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,14 @@ import (

sentryhttp "github.com/getsentry/sentry-go/http"

"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/rest"

"truefoundry/elasti/operator/internal/crddirectory"
"truefoundry/elasti/operator/internal/prom"

"sigs.k8s.io/controller-runtime/pkg/reconcile"

"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/truefoundry/elasti/pkg/k8shelper"
"github.com/truefoundry/elasti/pkg/messages"
Expand All @@ -39,17 +42,19 @@ type (
scaleLocks sync.Map
// rescaleDuration is the duration to wait before checking to rescaling the target
rescaleDuration time.Duration
reconciler reconcile.Reconciler
}
)

func NewServer(logger *zap.Logger, config *rest.Config, rescaleDuration time.Duration) *Server {
func NewServer(logger *zap.Logger, config *rest.Config, rescaleDuration time.Duration, reconciler reconcile.Reconciler) *Server {
// Get Ops client
k8sUtil := k8shelper.NewOps(logger, config)
return &Server{
logger: logger.Named("elastiServer"),
k8shelper: k8sUtil,
// rescaleDuration is the duration to wait before checking to rescaling the target
rescaleDuration: rescaleDuration,
reconciler: reconciler,
}
}

Expand Down Expand Up @@ -148,7 +153,7 @@ func (s *Server) resolverReqHandler(w http.ResponseWriter, req *http.Request) {
zap.String("namespace", body.Namespace))
}

func (s *Server) scaleTargetForService(_ context.Context, serviceName, namespace string) error {
func (s *Server) scaleTargetForService(ctx context.Context, serviceName, namespace string) error {
scaleMutex, loaded := s.getMutexForServiceScale(serviceName)
if loaded {
s.logger.Debug("Scale target lock already exists", zap.String("service", serviceName))
Expand All @@ -157,10 +162,26 @@ func (s *Server) scaleTargetForService(_ context.Context, serviceName, namespace
scaleMutex.Lock()
defer s.logger.Debug("Scale target lock released", zap.String("service", serviceName))
s.logger.Debug("Scale target lock taken", zap.String("service", serviceName))

crd, found := crddirectory.CRDDirectory.GetCRD(serviceName)
if !found {
s.releaseMutexForServiceScale(serviceName)
return fmt.Errorf("scaleTargetForService - error: failed to get CRD details from directory, serviceName: %s", serviceName)
s.logger.Debug("Failed to get CRD details from directory, running reconcile")
_, err := s.reconciler.Reconcile(ctx, reconcile.Request{
NamespacedName: types.NamespacedName{
Name: serviceName,
Namespace: namespace,
},
})
if err != nil {
s.releaseMutexForServiceScale(serviceName)
return fmt.Errorf("scaleTargetForService - error: reconcile failed, serviceName: %s, %w", serviceName, err)
}

crd, found = crddirectory.CRDDirectory.GetCRD(serviceName)
if !found {
s.releaseMutexForServiceScale(serviceName)
return fmt.Errorf("scaleTargetForService - error: failed to get CRD details from directory even after reconcile, serviceName: %s", serviceName)
}
}

if err := s.k8shelper.ScaleTargetWhenAtZero(namespace, crd.Spec.ScaleTargetRef.Name, crd.Spec.ScaleTargetRef.Kind, crd.Spec.MinTargetReplicas); err != nil {
Expand Down

0 comments on commit 77fa407

Please sign in to comment.