-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Create an enduro-am-worker and add an "am" Temporal Taskqueue - Branch the processing workflow logic for AM processing - Add a zip package - Add a ZipActivity to zip the bundled transfer before transmission - Add an UploadTransferActivity to upload the zipped transfer to the AM Storage Service transfer source directory - Add SFTP configuration for the upload - Add an SSH username and a logger to the SFTP package - Add the enduro-am-worker build to the Dockerfile - Add an enduro-am-worker to the Tilt dev deployment - Add a Tilt environment variable to select a kube config on startup - Add the enduro-am-worker container and configuration to the kube manifests - Add a "dev-am" kube overlay Co-authored-by: Diogenesoftoronto <[email protected]>
- Loading branch information
1 parent
e68a7c7
commit c9ee533
Showing
38 changed files
with
1,042 additions
and
91 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,3 +2,4 @@ | |
| /covreport | ||
| /dist | ||
| /.tilt.env | ||
| /*.secret | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,224 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "net/http" | ||
| "net/http/pprof" | ||
| "os" | ||
| "os/signal" | ||
| "syscall" | ||
| "time" | ||
|
|
||
| "github.com/oklog/run" | ||
| "github.com/prometheus/client_golang/prometheus/promhttp" | ||
| "github.com/spf13/pflag" | ||
| "go.artefactual.dev/tools/log" | ||
| temporalsdk_activity "go.temporal.io/sdk/activity" | ||
| temporalsdk_client "go.temporal.io/sdk/client" | ||
| temporalsdk_worker "go.temporal.io/sdk/worker" | ||
|
|
||
| "github.com/artefactual-sdps/enduro/internal/am" | ||
| "github.com/artefactual-sdps/enduro/internal/config" | ||
| "github.com/artefactual-sdps/enduro/internal/db" | ||
| "github.com/artefactual-sdps/enduro/internal/sftp" | ||
| "github.com/artefactual-sdps/enduro/internal/temporal" | ||
| "github.com/artefactual-sdps/enduro/internal/version" | ||
| "github.com/artefactual-sdps/enduro/internal/watcher" | ||
| "github.com/artefactual-sdps/enduro/internal/workflow/activities" | ||
| ) | ||
|
|
||
| const ( | ||
| appName = "enduro-am-worker" | ||
| ) | ||
|
|
||
| func main() { | ||
| p := pflag.NewFlagSet(appName, pflag.ExitOnError) | ||
|
|
||
| p.String("config", "", "Configuration file") | ||
| p.Bool("version", false, "Show version information") | ||
| _ = p.Parse(os.Args[1:]) | ||
|
|
||
| if v, _ := p.GetBool("version"); v { | ||
| fmt.Println(version.Info(appName)) | ||
| os.Exit(0) | ||
| } | ||
|
|
||
| var cfg config.Configuration | ||
| configFile, _ := p.GetString("config") | ||
| configFileFound, configFileUsed, err := config.Read(&cfg, configFile) | ||
| if err != nil { | ||
| fmt.Printf("Failed to read configuration: %v\n", err) | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| logger := log.New(os.Stderr, | ||
| log.WithName(appName), | ||
| log.WithDebug(cfg.Debug), | ||
| log.WithLevel(cfg.Verbosity), | ||
| ) | ||
| defer log.Sync(logger) | ||
|
|
||
| logger.Info("Starting...", "version", version.Long, "pid", os.Getpid()) | ||
|
|
||
| if configFileFound { | ||
| logger.Info("Configuration file loaded.", "path", configFileUsed) | ||
| } else { | ||
| logger.Info("Configuration file not found.") | ||
| } | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
|
|
||
| enduroDatabase, err := db.Connect(cfg.Database.Driver, cfg.Database.DSN) | ||
| if err != nil { | ||
| logger.Error(err, "Enduro database configuration failed.") | ||
| os.Exit(1) | ||
| } | ||
| _ = enduroDatabase.Ping() | ||
|
|
||
| temporalClient, err := temporalsdk_client.Dial(temporalsdk_client.Options{ | ||
| Namespace: cfg.Temporal.Namespace, | ||
| HostPort: cfg.Temporal.Address, | ||
| Logger: temporal.Logger(logger.WithName("temporal-client")), | ||
| }) | ||
| if err != nil { | ||
| logger.Error(err, "Error creating Temporal client.") | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| // Set up the watcher service. | ||
| var wsvc watcher.Service | ||
| { | ||
| wsvc, err = watcher.New(ctx, logger.WithName("watcher"), &cfg.Watcher) | ||
| if err != nil { | ||
| logger.Error(err, "Error setting up watchers.") | ||
| os.Exit(1) | ||
| } | ||
| } | ||
|
|
||
| var g run.Group | ||
|
|
||
| // Activity worker. | ||
| { | ||
| done := make(chan struct{}) | ||
| workerOpts := temporalsdk_worker.Options{ | ||
| DisableWorkflowWorker: true, | ||
| EnableSessionWorker: true, | ||
| MaxConcurrentSessionExecutionSize: 1000, | ||
| MaxConcurrentActivityExecutionSize: 1, | ||
| } | ||
| w := temporalsdk_worker.New(temporalClient, temporal.AmWorkerTaskQueue, workerOpts) | ||
| if err != nil { | ||
| logger.Error(err, "Error creating Temporal worker.") | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| w.RegisterActivityWithOptions( | ||
| activities.NewDownloadActivity(wsvc).Execute, | ||
| temporalsdk_activity.RegisterOptions{Name: activities.DownloadActivityName}, | ||
| ) | ||
| w.RegisterActivityWithOptions( | ||
| activities.NewBundleActivity(wsvc).Execute, | ||
| temporalsdk_activity.RegisterOptions{Name: activities.BundleActivityName}, | ||
| ) | ||
| w.RegisterActivityWithOptions( | ||
| activities.NewZipActivity(logger).Execute, temporalsdk_activity.RegisterOptions{Name: activities.ZipActivityName}, | ||
| ) | ||
| w.RegisterActivityWithOptions( | ||
| am.NewUploadTransferActivity(logger, sftp.NewGoClient(logger, cfg.AM.SFTP)).Execute, | ||
| temporalsdk_activity.RegisterOptions{Name: am.UploadTransferActivityName}, | ||
| ) | ||
| w.RegisterActivityWithOptions(activities.NewCleanUpActivity().Execute, temporalsdk_activity.RegisterOptions{Name: activities.CleanUpActivityName}) | ||
|
|
||
| g.Add( | ||
| func() error { | ||
| if err := w.Start(); err != nil { | ||
| return err | ||
| } | ||
| <-done | ||
| return nil | ||
| }, | ||
| func(err error) { | ||
| w.Stop() | ||
| close(done) | ||
| }, | ||
| ) | ||
| } | ||
|
|
||
| // Observability server. | ||
| { | ||
| srv := &http.Server{ | ||
| Addr: cfg.DebugListen, | ||
| ReadTimeout: time.Second * 1, | ||
| WriteTimeout: time.Second * 1, | ||
| IdleTimeout: time.Second * 30, | ||
| } | ||
|
|
||
| g.Add(func() error { | ||
| mux := http.NewServeMux() | ||
|
|
||
| // Health check. | ||
| mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { | ||
| w.WriteHeader(http.StatusOK) | ||
| fmt.Fprintln(w, "OK") | ||
| }) | ||
|
|
||
| // Prometheus metrics. | ||
| mux.Handle("/metrics", promhttp.Handler()) | ||
|
|
||
| // Profiling data. | ||
| mux.HandleFunc("/debug/pprof/", pprof.Index) | ||
| mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) | ||
| mux.HandleFunc("/debug/pprof/profile", pprof.Profile) | ||
| mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) | ||
| mux.HandleFunc("/debug/pprof/trace", pprof.Trace) | ||
| mux.Handle("/debug/pprof/block", pprof.Handler("block")) | ||
| mux.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine")) | ||
| mux.Handle("/debug/pprof/heap", pprof.Handler("heap")) | ||
| mux.Handle("/debug/pprof/threadcreate", pprof.Handler("threadcreate")) | ||
|
|
||
| srv.Handler = mux | ||
|
|
||
| return srv.ListenAndServe() | ||
| }, func(error) { | ||
| ctx, cancel := context.WithTimeout(ctx, time.Second*5) | ||
| defer cancel() | ||
| _ = srv.Shutdown(ctx) | ||
| }) | ||
| } | ||
|
|
||
| // Signal handler. | ||
| { | ||
| var ( | ||
| cancelInterrupt = make(chan struct{}) | ||
| ch = make(chan os.Signal, 2) | ||
| ) | ||
| defer close(ch) | ||
|
|
||
| g.Add( | ||
| func() error { | ||
| signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM) | ||
|
|
||
| select { | ||
| case <-ch: | ||
| case <-cancelInterrupt: | ||
| } | ||
|
|
||
| return nil | ||
| }, func(err error) { | ||
| logger.Info("Quitting...") | ||
| close(cancelInterrupt) | ||
| cancel() | ||
| signal.Stop(ch) | ||
| }, | ||
| ) | ||
| } | ||
|
|
||
| err = g.Run() | ||
| if err != nil { | ||
| logger.Error(err, "Application failure.") | ||
| os.Exit(1) | ||
| } | ||
| logger.Info("Bye!") | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.