Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
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
7 changes: 7 additions & 0 deletions internal/config/providers.go
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,9 @@ type initResult struct {

// InitializeProviders initializes all providers in parallel using channels
func (c *Config) InitializeProviders() error {
if c.IsServer() {
_ = c.GetServices()
Comment thread
hughneale marked this conversation as resolved.
}

defs := c.GetProviders().Definitions

Expand Down Expand Up @@ -365,6 +368,10 @@ func (c *Config) InitializeProviders() error {
c.providerInstances = results
c.mu.Unlock()

if err := c.StartTemporalWorkers(); err != nil {
return err
}

logrus.Debugln("All providers initialized successfully")

return nil
Expand Down
32 changes: 23 additions & 9 deletions internal/config/services.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,22 +18,20 @@ func (c *Config) GetServices() models.ServicesClientImpl {
return
}
c.servicesClient = newClient
go func() {
// Post services setup initialization, we need to do some additional setup for certain services that are dependent on the configuration being fully loaded.
err = c.SetupTemporal()
if err != nil {
logrus.WithError(err).Error("Failed to set up temporal services")
}
}()

// Post services setup initialization for services that depend on full config state.
err = c.SetupTemporal()
if err != nil {
logrus.WithError(err).Error("Failed to set up temporal services")
}
Comment thread
hughneale marked this conversation as resolved.
Outdated
})

return c.servicesClient

}

func (c *Config) SetupTemporal() error {

if c.GetServices() != nil && c.GetServices().GetTemporal() != nil {
if c.servicesClient != nil && c.servicesClient.GetTemporal() != nil {

logrus.Infoln("Setting up temporal services...")

Expand All @@ -60,3 +58,19 @@ func (c *Config) SetupTemporal() error {
return nil

}

func (c *Config) StartTemporalWorkers() error {
if c.servicesClient == nil || c.servicesClient.GetTemporal() == nil {
return nil
}

if !c.IsServer() {
return nil
}

if err := c.servicesClient.GetTemporal().StartWorkers(); err != nil {
return fmt.Errorf("starting temporal workers: %w", err)
}

return nil
}
61 changes: 52 additions & 9 deletions internal/config/services/temporal/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,12 +27,14 @@ type TemporalClient struct {
config *models.TemporalConfig
client client.Client
workers map[string]worker.Worker
started map[string]worker.Worker
identities []string
Comment thread
hughneale marked this conversation as resolved.
vault models.VaultImpl

mu sync.Mutex
readyCh chan struct{}
closeReadyOnce sync.Once
workersStarted bool
}

func NewTemporalClient(
Expand All @@ -57,6 +59,7 @@ func NewTemporalClient(
identities: unique,
vault: vault,
workers: make(map[string]worker.Worker, len(unique)),
started: make(map[string]worker.Worker, len(unique)),
readyCh: make(chan struct{}),
}
}
Expand Down Expand Up @@ -131,7 +134,8 @@ func (a *TemporalClient) Initialize() error {
}
}

// Create and start a worker for each identity (task queue)
// Create a worker for each identity (task queue).
// Registration must happen before workers are started.
a.mu.Lock()
defer a.mu.Unlock()

Expand All @@ -147,32 +151,69 @@ func (a *TemporalClient) Initialize() error {
workerOptions,
)

a.workers[identity] = newWorker
}

if len(a.workers) == 0 {
a.markReady() // Unblock any waiters even on failure
return fmt.Errorf("failed to create any Temporal workers")
}

return nil
}

// StartWorkers starts all registered Temporal workers.
// This must be called only after workflow/activity registration is complete.
func (c *TemporalClient) StartWorkers() error {
c.mu.Lock()
defer c.mu.Unlock()

if c.client == nil {
return fmt.Errorf("temporal client is not initialized")
}

if len(c.workers) == 0 {
c.markReady()
return fmt.Errorf("no Temporal workers configured")
}

if c.workersStarted {
logrus.Warn("Temporal workers already started, skipping worker startup")
return nil
}

buildID := common.GetBuildIdentifier()

for identity, w := range c.workers {
logrus.WithFields(logrus.Fields{
"BuildID": buildID,
"taskQueue": identity,
}).Infof("Starting Temporal worker")
}).Info("Starting Temporal worker")

if err := newWorker.Start(); err != nil {
if err := w.Start(); err != nil {
logrus.WithError(err).
WithField("taskQueue", identity).
Error("Failed to start temporal worker")
delete(c.workers, identity)
continue
}

a.workers[identity] = newWorker
c.started[identity] = w
}

if len(a.workers) == 0 {
a.markReady() // Unblock any waiters even on failure
if len(c.started) == 0 {
c.markReady()
return fmt.Errorf("failed to start any Temporal workers")
}

c.workersStarted = true

// If versioning is enabled, confirm our deployment version is registered
// on the Temporal server before allowing workflow submissions via GetClient().
if a.config.DisableVersioning {
a.markReady()
if c.config.DisableVersioning {
c.markReady()
} else {
go a.awaitVersionRegistration(buildID)
go c.awaitVersionRegistration(buildID)
}

return nil
Expand Down Expand Up @@ -334,7 +375,9 @@ func (c *TemporalClient) Shutdown() error {
}

c.workers = nil
c.started = nil
c.client = nil
c.workersStarted = false

return nil
}
Expand Down
2 changes: 1 addition & 1 deletion internal/config/services/temporal/multi_worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import (
// identity-specific task queues.
//
// Lifecycle methods (Start/Run/Stop) are no-ops because TemporalClient
// manages worker lifecycle directly in Initialize() and Shutdown().
// manages worker lifecycle directly via StartWorkers() and Shutdown().
type multiWorker struct {
workers []worker.Worker
}
Expand Down
6 changes: 2 additions & 4 deletions internal/config/temporal.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,7 @@ import (

// Register temporal workflows and activities
func (c *Config) registerTemporalWorkflows() error {

if c.GetServices() == nil || c.GetServices().GetTemporal() == nil {
if c.servicesClient == nil || c.servicesClient.GetTemporal() == nil {
return fmt.Errorf("temporal service is not initialized")
}

Expand All @@ -26,8 +25,7 @@ func (c *Config) registerTemporalWorkflows() error {
}

func (c *Config) registerTemporalActivities() error {

if c.GetServices() == nil || c.GetServices().GetTemporal() == nil {
if c.servicesClient == nil || c.servicesClient.GetTemporal() == nil {
return fmt.Errorf("temporal service is not initialized")
}

Expand Down
1 change: 1 addition & 0 deletions internal/models/temporal.go
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ func (t *TemporalConfig) HasMtlsConfig() bool {

type TemporalImpl interface {
Initialize() error
StartWorkers() error
Shutdown() error

GetClient() client.Client
Expand Down
1 change: 1 addition & 0 deletions test/integration/services/temporal_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ func initAndRegister(t *testing.T, infra *testinfra.TestInfrastructure, cfg *mod
require.NotNil(t, w, "GetWorker must return a worker after Initialize")
w.RegisterWorkflow(echoWorkflow)
w.RegisterActivity(echoActivity)
require.NoError(t, tc.StartWorkers(), "StartWorkers should succeed")
return tc
}

Expand Down
Loading