Add temporal GetSystemInfo timeout config (#838)

Co-authored-by: Dev Agent <dev-agent@example.com>

Author: pulltheflower

cmd/csghub-server/cmd/moderation/launch.go

@@ -1,16 +1,22 @@
 package moderation
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
+	"time"
+
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/log"
+	"opencsg.com/csghub-server/builder/instrumentation"
+	"opencsg.com/csghub-server/builder/temporal"
 
 	"github.com/spf13/cobra"
 	"opencsg.com/csghub-server/api/httpbase"
 	"opencsg.com/csghub-server/builder/store/database"
 	"opencsg.com/csghub-server/common/config"
 	"opencsg.com/csghub-server/moderation/checker"
 	"opencsg.com/csghub-server/moderation/router"
-	"opencsg.com/csghub-server/moderation/workflow"
 )
 
 var cmdLaunch = &cobra.Command{
@@ -23,6 +29,10 @@ var cmdLaunch = &cobra.Command{
 			return err
 		}
 		slog.Debug("config", slog.Any("data", cfg))
+		stopOtel, err := instrumentation.SetupOTelSDK(context.Background(), cfg, instrumentation.Moderation)
+		if err != nil {
+			panic(err)
+		}
 		// Check APIToken length
 		if len(cfg.APIToken) < 128 {
 			return fmt.Errorf("API token length is less than 128, please check")
@@ -36,11 +46,16 @@ var cmdLaunch = &cobra.Command{
 			return fmt.Errorf("database initialization failed: %w", err)
 		}
 		checker.Init(cfg)
-
-		//init async moderation process
-		err = workflow.StartWorker(cfg)
+		slog.Info("starting temporal client")
+		temporalClient, err := temporal.NewClient(client.Options{
+			HostPort: cfg.WorkFLow.Endpoint,
+			Logger:   log.NewStructuredLogger(slog.Default()),
+			ConnectionOptions: client.ConnectionOptions{
+				GetSystemInfoTimeout: time.Duration(cfg.Temporal.GetSystemInfoTimeout) * time.Second,
+			},
+		}, instrumentation.Moderation)
 		if err != nil {
-			return fmt.Errorf("failed to start workflow worker,%w", err)
+			return fmt.Errorf("unable to create temporal client, error: %w", err)
 		}
 
 		r, err := router.NewRouter(cfg)
@@ -56,8 +71,8 @@ var cmdLaunch = &cobra.Command{
 		)
 		server.Run()
 
-		workflow.StopWorker()
-
+		_ = stopOtel(context.Background())
+		temporalClient.Close()
 		return nil
 	},
 }

cmd/csghub-server/cmd/notification/launch.go

@@ -1,15 +1,21 @@
 package notification
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
+	"time"
+
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/log"
+	"opencsg.com/csghub-server/builder/instrumentation"
+	"opencsg.com/csghub-server/builder/temporal"
 
 	"github.com/spf13/cobra"
 	"opencsg.com/csghub-server/api/httpbase"
 	"opencsg.com/csghub-server/builder/store/database"
 	"opencsg.com/csghub-server/common/config"
 	"opencsg.com/csghub-server/notification/router"
-	"opencsg.com/csghub-server/notification/workflow"
 )
 
 var launchCmd = &cobra.Command{
@@ -21,7 +27,10 @@ var launchCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
-
+		stopOtel, err := instrumentation.SetupOTelSDK(context.Background(), cfg, instrumentation.Notification)
+		if err != nil {
+			panic(err)
+		}
 		// Check APIToken length
 		if len(cfg.APIToken) < 128 {
 			return fmt.Errorf("API token length is less than 128, please check")
@@ -34,10 +43,15 @@ var launchCmd = &cobra.Command{
 			slog.Error("failed to initialize database", slog.Any("error", err))
 			return fmt.Errorf("database initialization failed: %w", err)
 		}
-
-		err = workflow.StartWorkflow(cfg)
+		workflowClient, err := temporal.NewClient(client.Options{
+			HostPort: cfg.WorkFLow.Endpoint,
+			Logger:   log.NewStructuredLogger(slog.Default()),
+			ConnectionOptions: client.ConnectionOptions{
+				GetSystemInfoTimeout: time.Duration(cfg.Temporal.GetSystemInfoTimeout) * time.Second,
+			},
+		}, "csghub-notification")
 		if err != nil {
-			slog.Warn("failed to start workflow", slog.Any("error", err))
+			return fmt.Errorf("unable to create workflow client, error: %w", err)
 		}
 
 		r, err := router.NewNotifierRouter(cfg)
@@ -53,6 +67,9 @@ var launchCmd = &cobra.Command{
 		)
 		server.Run()
 
+		_ = stopOtel(context.Background())
+		workflowClient.Close()
+
 		return nil
 	},
 }

cmd/csghub-server/cmd/temporal-worker/launch.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"time"
 
 	"opencsg.com/csghub-server/moderation/checker"
 
@@ -81,6 +82,9 @@ var cmdLaunch = &cobra.Command{
 		temporalClient, err := temporal.NewClient(client.Options{
 			HostPort: cfg.WorkFLow.Endpoint,
 			Logger:   log.NewStructuredLogger(slog.Default()),
+			ConnectionOptions: client.ConnectionOptions{
+				GetSystemInfoTimeout: time.Duration(cfg.Temporal.GetSystemInfoTimeout) * time.Second,
+			},
 		}, instrumentation.TemporalWorker)
 		if err != nil {
 			return fmt.Errorf("unable to create temporal client, error: %w", err)

cmd/csghub-server/cmd/user/launch.go

@@ -1,15 +1,21 @@
 package user
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
+	"time"
+
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/log"
+	"opencsg.com/csghub-server/builder/instrumentation"
+	"opencsg.com/csghub-server/builder/temporal"
 
 	"github.com/spf13/cobra"
 	"opencsg.com/csghub-server/api/httpbase"
 	"opencsg.com/csghub-server/builder/store/database"
 	"opencsg.com/csghub-server/common/config"
 	"opencsg.com/csghub-server/user/router"
-	"opencsg.com/csghub-server/user/workflow"
 )
 
 var cmdLaunch = &cobra.Command{
@@ -22,6 +28,10 @@ var cmdLaunch = &cobra.Command{
 			return err
 		}
 		slog.Debug("config", slog.Any("data", cfg))
+		stopOtel, err := instrumentation.SetupOTelSDK(context.Background(), cfg, instrumentation.User)
+		if err != nil {
+			panic(err)
+		}
 		// Check APIToken length
 		if len(cfg.APIToken) < 128 {
 			return fmt.Errorf("API token length is less than 128, please check")
@@ -35,9 +45,15 @@ var cmdLaunch = &cobra.Command{
 			return fmt.Errorf("database initialization failed: %w", err)
 		}
 
-		err = workflow.StartWorker(cfg)
+		wfClient, err := temporal.NewClient(client.Options{
+			HostPort: cfg.WorkFLow.Endpoint,
+			Logger:   log.NewStructuredLogger(slog.Default()),
+			ConnectionOptions: client.ConnectionOptions{
+				GetSystemInfoTimeout: time.Duration(cfg.Temporal.GetSystemInfoTimeout) * time.Second,
+			},
+		}, "csghub-user")
 		if err != nil {
-			return fmt.Errorf("failed to start user workflow worker: %w", err)
+			return fmt.Errorf("unable to create workflow client, error:%w", err)
 		}
 
 		r, err := router.NewRouter(cfg)
@@ -53,8 +69,8 @@ var cmdLaunch = &cobra.Command{
 		)
 		server.Run()
 
-		workflow.StopWorker()
-
+		_ = stopOtel(context.Background())
+		wfClient.Close()
 		return nil
 	},
 }