From 94f50ce6e14ca1a5fc9a9ad4e8283c87cb02e47d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 20 Jul 2026 12:01:10 +0000 Subject: [PATCH] fix(scheduler,cli): bug-1453/1454/1455/1456 remove dead CA-cert plumbing, fix scheduler HA config bugs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bug-1453: remove unused client-side CA-certificate plumbing. C-side: drop ca_cert_path from xce_cluster_config_t, do_connect()'s ca_cert param, --ca-cert flags on worker/scheduler, and controller's cert population (TLS is server-side only, SSL_VERIFY_NONE everywhere on the client side already, verified zero SSL_CTX_load_verify_locations calls in the C tree). Go-side: delete xcecli's self-signed cert TOFU-pinning flow (cmd/auth/cert/*), CACert field, SetRootCAs/ loadRootCAs, and every call site. bug-1454: fix xcecli_ctx_t leak in scheduler.c sched_init() — the unconditional initial ctx_init was never freed before the HA candidate loop re-initialized it on the first iteration. bug-1455: sched_init()'s memset(sched, 0, ...) was silently wiping sched->cluster_cfg that main.c had populated from --peers just before calling sched_init(), so --peers was always ignored and the scheduler fell back to the hardcoded localhost:9990 single-host default. Restructured sched_init() to take cluster_cfg as an explicit parameter (assigned after the memset) instead of relying on caller-side ordering with a separate sched_set_cluster_cfg() setter, which is now removed. bug-1456: revert an unsafe stack-local xce_client_config_t introduced while fixing bug-1455 — xcecli_ctx_init() stores that pointer as ctx->conf and client_failover.c dereferences it for the ctx's entire life (runtime reconnect), so a per-loop-iteration stack copy would dangle the moment sched_init() returned. Back to mutating the caller-owned, process-lifetime config in place. Verified: full ASAN build (all components), meson test suite 9/9 OK, manual scheduler testsuite 15/15, Go build/vet/test all pass. --- src/controller/main.c | 4 - src/frontend/xcecli/cmd/auth/cert/cert.go | 82 -- src/frontend/xcecli/cmd/auth/cert/parser.go | 37 - src/frontend/xcecli/cmd/auth/logon.go | 360 ++++----- src/frontend/xcecli/cmd/clusters/list.go | 159 ++-- src/frontend/xcecli/cmd/hosts/controller.go | 356 +++++---- src/frontend/xcecli/cmd/hosts/list.go | 397 +++++----- src/frontend/xcecli/cmd/hosts/select.go | 3 +- src/frontend/xcecli/cmd/jobs/controller.go | 572 +++++++------- src/frontend/xcecli/cmd/jobs/list.go | 737 +++++++++--------- src/frontend/xcecli/cmd/jobs/restart.go | 10 +- src/frontend/xcecli/cmd/jobs/submit.go | 3 - src/frontend/xcecli/cmd/permissions/list.go | 3 +- src/frontend/xcecli/cmd/queues/controller.go | 476 ++++++----- src/frontend/xcecli/cmd/queues/list.go | 397 +++++----- src/frontend/xcecli/cmd/resources/list.go | 6 +- src/frontend/xcecli/cmd/roles/addrole.go | 4 - src/frontend/xcecli/cmd/roles/deleterole.go | 2 - src/frontend/xcecli/cmd/roles/list.go | 3 +- src/frontend/xcecli/cmd/users/assignrole.go | 2 - src/frontend/xcecli/cmd/users/list.go | 6 +- src/frontend/xcecli/cmd/users/perm4user.go | 3 +- src/frontend/xcecli/cmd/users/unassignrole.go | 2 - src/frontend/xcecli/common/request/request.go | 51 +- .../xcecli/common/request/request_test.go | 30 - src/frontend/xcecli/config/config.go | 407 +++++----- src/frontend/xcecli/models/logon.go | 37 +- src/include/xce/cluster.h | 14 +- src/lib/cluster.c | 10 +- src/lib/testsuite/test_cluster.c | 5 +- src/scheduler/include/scheduler.h | 3 +- src/scheduler/src/main.c | 34 +- src/scheduler/src/scheduler.c | 26 +- src/worker/src/main.c | 32 +- 34 files changed, 1999 insertions(+), 2274 deletions(-) delete mode 100755 src/frontend/xcecli/cmd/auth/cert/cert.go delete mode 100755 src/frontend/xcecli/cmd/auth/cert/parser.go diff --git a/src/controller/main.c b/src/controller/main.c index b2c502b0..d2062d15 100644 --- a/src/controller/main.c +++ b/src/controller/main.c @@ -275,10 +275,6 @@ int main(int argc, char* argv[]) return 1; } - /* CA cert from certdir (same certificate as certfile) */ - snprintf(myconf.peers_cfg.ca_cert_path, - sizeof(myconf.peers_cfg.ca_cert_path), - "%s/xce.crt", certdir); strncpy(myconf.peers_cfg.identity, arguments.myname, sizeof(myconf.peers_cfg.identity) - 1); INFO("HA mode: %d controller candidates, identity=%s", diff --git a/src/frontend/xcecli/cmd/auth/cert/cert.go b/src/frontend/xcecli/cmd/auth/cert/cert.go deleted file mode 100755 index a9d030e1..00000000 --- a/src/frontend/xcecli/cmd/auth/cert/cert.go +++ /dev/null @@ -1,82 +0,0 @@ -package cert - -import ( - "crypto/tls" - "encoding/pem" - "log" - "os" - "os/user" - "path" - logger "xcecli/config" - - "go.uber.org/zap" -) - -func GeneratorCert(serverAddress, configPath string) error { - serverAddress, err := extractHostAndPort(serverAddress) - logger.Debug("Server address", zap.String("serverAddress", serverAddress)) - if err != nil { - return err - } - // The HTTPS server's address (replace with the server you want to check) - //serverAddress := "172.16.10.10:8443" - - // Create a tls.Config with InsecureSkipVerify set to true - tlsConfig := &tls.Config{ - InsecureSkipVerify: true, - } - - // Dial the server using TLS with the custom tls.Config - conn, err := tls.Dial("tcp", serverAddress, tlsConfig) - if err != nil { - log.Fatal(err) - } - defer conn.Close() - - // Retrieve the certificate chain - certs := conn.ConnectionState().PeerCertificates - - // Print the certificate chain - // for i, cert := range certs { - // fmt.Printf("Certificate %d:\n", i) - // fmt.Printf(" Subject: %s\n", cert.Subject) - // fmt.Printf(" Issuer: %s\n", cert.Issuer) - // fmt.Printf(" Not Before: %s\n", cert.NotBefore) - // fmt.Printf(" Not After: %s\n", cert.NotAfter) - // } - - // The root CA is usually the last certificate in the chain - //rootCert := certs[len(certs)-1] - // fmt.Println("\nRoot CA Certificate:") - // fmt.Printf(" Subject: %s\n", rootCert.Subject) - // fmt.Printf(" Issuer: %s\n", rootCert.Issuer) - // fmt.Printf(" Not Before: %s\n", rootCert.NotBefore) - // fmt.Printf(" Not After: %s\n", rootCert.NotAfter) - - // Save the server certificate to a file - certOut, err := os.Create(configPath) - if err != nil { - return err - } - defer certOut.Close() - - pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: certs[0].Raw}) - //fmt.Println("Server certificate saved to server.crt") - return nil -} - -func GetCertPath() (string, error) { - usr, err := user.Current() - if err != nil { - return "", err - } - - configDir := path.Join(usr.HomeDir, ".xce") - configPath := path.Join(configDir, "server.crt") - - // create the config directory if it doesn't exist - if err := os.MkdirAll(configDir, 0700); err != nil { - return "", err - } - return configPath, nil -} diff --git a/src/frontend/xcecli/cmd/auth/cert/parser.go b/src/frontend/xcecli/cmd/auth/cert/parser.go deleted file mode 100755 index 8285c8bd..00000000 --- a/src/frontend/xcecli/cmd/auth/cert/parser.go +++ /dev/null @@ -1,37 +0,0 @@ -package cert - -import ( - "fmt" - "net/url" - "xcecli/common/request" -) - -func extractHostAndPort(rawURL string) (string, error) { - // parse the URL - parsedURL, err := url.Parse(rawURL) - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return "", fmt.Errorf("failed to parse URL: [%s]", errMsg) - } - - // get the host and port - host := parsedURL.Hostname() - port := parsedURL.Port() - - // if the port is empty, use the default port (HTTP is 80, HTTPS is 443) - if port == "" { - if parsedURL.Scheme == "https" { - port = "443" - } else if parsedURL.Scheme == "http" { - port = "80" - } - } - - // concatenate the host and port - result := host - if port != "" { - result = fmt.Sprintf("%s:%s", host, port) - } - - return result, nil -} diff --git a/src/frontend/xcecli/cmd/auth/logon.go b/src/frontend/xcecli/cmd/auth/logon.go index e2f59d14..1144f0ad 100755 --- a/src/frontend/xcecli/cmd/auth/logon.go +++ b/src/frontend/xcecli/cmd/auth/logon.go @@ -1,189 +1,171 @@ -package cmd - -import ( - "fmt" - "net/url" - "os" - "strings" - "syscall" - "xcecli/cmd/auth/cert" - "xcecli/common/request" - "xcecli/config" - logger "xcecli/config" - "xcecli/models" - - "go.uber.org/zap" - "golang.org/x/term" - - "github.com/spf13/cobra" -) - -// updateServerConfig -func updateServerConfig(cfg *config.Config, url, certPath string, loginResp *models.LogonResponse) error { - logger.Debug("updateServerConfig", zap.String("url", url), zap.String("certPath", certPath)) - logger.Info("updateServerConfig", zap.String("url", url), zap.String("certPath", certPath)) - - // check if the server with the same URL already exists - Found := false - for i, server := range cfg.APIServerInfo { - if server.URL == url { - // update the existing server info - cfg.APIServerInfo[i].Token = loginResp.Data.Token - cfg.APIServerInfo[i].Path = loginResp.Data.Path //save the path - cfg.APIServerInfo[i].CreateTime = loginResp.Data.CreateTime - cfg.APIServerInfo[i].CACert = certPath - Found = true - break - } - } - - // if it's a new server, add it to the list - if !Found { - - newServer := config.APIServerInfo{ - URL: url, - Token: loginResp.Data.Token, - Path: loginResp.Data.Path, - CreateTime: loginResp.Data.CreateTime, - CACert: certPath, - } - - // add it to the end of the array - cfg.APIServerInfo = append(cfg.APIServerInfo, newServer) - logger.Debug("newServer", zap.Any("newServer", newServer)) - logger.Info("newServer", zap.Any("newServer", newServer)) - } - - return nil -} - -func NewLogonCmd(configManager *config.ConfigManager) *cobra.Command { - var opts struct { - username string - password string - url string - } - logger.Debug("NewLogonCmd", zap.Any("opts", opts)) - logger.Info("NewLogonCmd", zap.Any("opts", opts)) - - cmd := &cobra.Command{ - Use: "login", - GroupID: "api", - Short: "Login to API Server", - Long: `Login to the specified API Server, and save the login info to the config file. -Example: - xce login -u USERNAME -p PASSWORD --url http://IP:8080 - xce login -u USERNAME -p PASSWORD --url https://IP:8443`, - Run: func(cmd *cobra.Command, args []string) { - // if the password is not provided, prompt the user to enter it - if opts.password == "" { - fmt.Print("enter password: ") - bytePassword, err := term.ReadPassword(int(syscall.Stdin)) - fmt.Println() // new line - if err != nil { - errMsg := request.ExtractErrorMessage(err) - fmt.Printf("error reading password: [%s]", errMsg) - os.Exit(1) - } - opts.password = string(bytePassword) - } - - err := runLogon(opts, configManager) - if err != nil { - fmt.Println(strings.ToLower(err.Error())) - } - }, - } - - // set the command line parameters - flags := cmd.Flags() - flags.StringVarP(&opts.username, "username", "u", "", "username") - flags.StringVarP(&opts.password, "password", "p", "", "password (optional, will prompt if not provided)") - flags.StringVar(&opts.url, "url", "", "API Server URL") - - // required parameters - cmd.MarkFlagRequired("username") - - //cmd.MarkFlagRequired("password") - cmd.MarkFlagRequired("url") - - logger.Debug("cmd", zap.Any("cmd", cmd)) - logger.Info("cmd", zap.Any("cmd", cmd)) - - return cmd -} - -func runLogon(opts struct{ username, password, url string }, cm *config.ConfigManager) error { - if cm == nil { - return fmt.Errorf("fail to initialize the config manager") - } - logger.Debug("runLogon", zap.Any("opts", opts)) - logger.Info("runLogon", zap.Any("opts", opts)) - - //get the config - cfg, err := cm.GetConfig() - if err != nil { - cfg = &config.Config{} - } - logger.Debug("cfg", zap.Any("cfg", cfg)) - logger.Info("cfg", zap.Any("cfg", cfg)) - - // create the API client - apiClient := request.NewAPIClient(opts.url) - certPath := "" - // Only setup certs for HTTPS connections - if u, err := url.Parse(opts.url); err == nil && u.Scheme == "https" { - certPath, err = cert.GetCertPath() - apiClient.SetRootCAs(certPath) - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to get cert path: [%s]", errMsg) - } - if err := cert.GeneratorCert(opts.url, certPath); err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to generate cert: [%s]", errMsg) - } - } - logger.Debug("apiClient", zap.Any("apiClient", apiClient)) - logger.Info("apiClient", zap.Any("apiClient", apiClient)) - - // perform the login - loginResp, err := apiClient.Logon(opts.username, opts.password) - if err != nil { - // If login fails, remove the server entry with the same URL if it exists - for i, server := range cfg.APIServerInfo { - if server.URL == opts.url { - // Remove this server from the slice - cfg.APIServerInfo = append(cfg.APIServerInfo[:i], cfg.APIServerInfo[i+1:]...) - // Save the updated config - if saveErr := cm.SaveConfig(cfg); saveErr != nil { - logger.Error("Failed to save the config after removing server entry", zap.Error(saveErr)) - } - break - } - } - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to login: [%s]", errMsg) - } - logger.Debug("loginResp", zap.Any("loginResp", loginResp)) - logger.Info("loginResp", zap.Any("loginResp", loginResp)) - - // update the server config - if err := updateServerConfig(cfg, opts.url, certPath, loginResp); err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to update the server config: [%s]", errMsg) - } - logger.Debug("cfg", zap.Any("cfg", cfg)) - logger.Info("cfg", zap.Any("cfg", cfg)) - - // save the config - if err := cm.SaveConfig(cfg); err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to save the config: [%s]", errMsg) - } - logger.Debug("cfg", zap.Any("cfg", cfg)) - logger.Info("cfg", zap.Any("cfg", cfg)) - - fmt.Println("login successfully") - return nil -} +package cmd + +import ( + "fmt" + "os" + "strings" + "syscall" + "xcecli/common/request" + "xcecli/config" + logger "xcecli/config" + "xcecli/models" + + "go.uber.org/zap" + "golang.org/x/term" + + "github.com/spf13/cobra" +) + +// updateServerConfig +func updateServerConfig(cfg *config.Config, url string, loginResp *models.LogonResponse) error { + logger.Debug("updateServerConfig", zap.String("url", url)) + logger.Info("updateServerConfig", zap.String("url", url)) + + // check if the server with the same URL already exists + Found := false + for i, server := range cfg.APIServerInfo { + if server.URL == url { + // update the existing server info + cfg.APIServerInfo[i].Token = loginResp.Data.Token + cfg.APIServerInfo[i].Path = loginResp.Data.Path //save the path + cfg.APIServerInfo[i].CreateTime = loginResp.Data.CreateTime + Found = true + break + } + } + + // if it's a new server, add it to the list + if !Found { + + newServer := config.APIServerInfo{ + URL: url, + Token: loginResp.Data.Token, + Path: loginResp.Data.Path, + CreateTime: loginResp.Data.CreateTime, + } + + // add it to the end of the array + cfg.APIServerInfo = append(cfg.APIServerInfo, newServer) + logger.Debug("newServer", zap.Any("newServer", newServer)) + logger.Info("newServer", zap.Any("newServer", newServer)) + } + + return nil +} + +func NewLogonCmd(configManager *config.ConfigManager) *cobra.Command { + var opts struct { + username string + password string + url string + } + logger.Debug("NewLogonCmd", zap.Any("opts", opts)) + logger.Info("NewLogonCmd", zap.Any("opts", opts)) + + cmd := &cobra.Command{ + Use: "login", + GroupID: "api", + Short: "Login to API Server", + Long: `Login to the specified API Server, and save the login info to the config file. +Example: + xce login -u USERNAME -p PASSWORD --url http://IP:8080 + xce login -u USERNAME -p PASSWORD --url https://IP:8443`, + Run: func(cmd *cobra.Command, args []string) { + // if the password is not provided, prompt the user to enter it + if opts.password == "" { + fmt.Print("enter password: ") + bytePassword, err := term.ReadPassword(int(syscall.Stdin)) + fmt.Println() // new line + if err != nil { + errMsg := request.ExtractErrorMessage(err) + fmt.Printf("error reading password: [%s]", errMsg) + os.Exit(1) + } + opts.password = string(bytePassword) + } + + err := runLogon(opts, configManager) + if err != nil { + fmt.Println(strings.ToLower(err.Error())) + } + }, + } + + // set the command line parameters + flags := cmd.Flags() + flags.StringVarP(&opts.username, "username", "u", "", "username") + flags.StringVarP(&opts.password, "password", "p", "", "password (optional, will prompt if not provided)") + flags.StringVar(&opts.url, "url", "", "API Server URL") + + // required parameters + cmd.MarkFlagRequired("username") + + //cmd.MarkFlagRequired("password") + cmd.MarkFlagRequired("url") + + logger.Debug("cmd", zap.Any("cmd", cmd)) + logger.Info("cmd", zap.Any("cmd", cmd)) + + return cmd +} + +func runLogon(opts struct{ username, password, url string }, cm *config.ConfigManager) error { + if cm == nil { + return fmt.Errorf("fail to initialize the config manager") + } + logger.Debug("runLogon", zap.Any("opts", opts)) + logger.Info("runLogon", zap.Any("opts", opts)) + + //get the config + cfg, err := cm.GetConfig() + if err != nil { + cfg = &config.Config{} + } + logger.Debug("cfg", zap.Any("cfg", cfg)) + logger.Info("cfg", zap.Any("cfg", cfg)) + + // create the API client + apiClient := request.NewAPIClient(opts.url) + logger.Debug("apiClient", zap.Any("apiClient", apiClient)) + logger.Info("apiClient", zap.Any("apiClient", apiClient)) + + // perform the login + loginResp, err := apiClient.Logon(opts.username, opts.password) + if err != nil { + // If login fails, remove the server entry with the same URL if it exists + for i, server := range cfg.APIServerInfo { + if server.URL == opts.url { + // Remove this server from the slice + cfg.APIServerInfo = append(cfg.APIServerInfo[:i], cfg.APIServerInfo[i+1:]...) + // Save the updated config + if saveErr := cm.SaveConfig(cfg); saveErr != nil { + logger.Error("Failed to save the config after removing server entry", zap.Error(saveErr)) + } + break + } + } + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to login: [%s]", errMsg) + } + logger.Debug("loginResp", zap.Any("loginResp", loginResp)) + logger.Info("loginResp", zap.Any("loginResp", loginResp)) + + // update the server config + if err := updateServerConfig(cfg, opts.url, loginResp); err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to update the server config: [%s]", errMsg) + } + logger.Debug("cfg", zap.Any("cfg", cfg)) + logger.Info("cfg", zap.Any("cfg", cfg)) + + // save the config + if err := cm.SaveConfig(cfg); err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to save the config: [%s]", errMsg) + } + logger.Debug("cfg", zap.Any("cfg", cfg)) + logger.Info("cfg", zap.Any("cfg", cfg)) + + fmt.Println("login successfully") + return nil +} diff --git a/src/frontend/xcecli/cmd/clusters/list.go b/src/frontend/xcecli/cmd/clusters/list.go index 07aeaceb..456e94bf 100755 --- a/src/frontend/xcecli/cmd/clusters/list.go +++ b/src/frontend/xcecli/cmd/clusters/list.go @@ -1,80 +1,79 @@ -package clusters - -import ( - "fmt" - "strings" - "xcecli/common/output" - "xcecli/common/request" - "xcecli/config" - "xcecli/models" - - "github.com/spf13/cobra" -) - -var ( - cluster string - fields string - Verbose bool -) - -// NewClustersCmd create cluster query command -func NewClustersCmd(configManager *config.ConfigManager) *cobra.Command { - cmd := &cobra.Command{ - Use: "id", - GroupID: "query", - Short: "Query cluster info", - Long: `Query cluster info, support filter by cluster name . -example: - xce clusters # query all clusters`, - - RunE: func(cmd *cobra.Command, args []string) error { - err := runClusters(configManager) - if err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - //cmd.Flags().BoolVarP(&Verbose, "verbose", "l", false, "show cluster detail") - return cmd -} - -func runClusters(cm *config.ConfigManager) error { - // get the config - serverInfo, _ := cm.GetServerInfo() - - // create API client - apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) - - // validate the parameters - cluster = strings.ToLower(cluster) - - // create the query parameters - params := make(map[string]string) - - // handle field selection - if fields != "" || cluster != "" { - params["fields"] = fields - } - - // create the API client and query the host info - var resp models.ClusterResponse - err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, "/xce/v1/clusters", &resp, params) - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to get clusters: [%s]", errMsg) - } - - if resp.Data.Count == 0 { - fmt.Println("no cluster found") - return nil - } - - // print result - output := output.NewOutput(fields, false, false) - output.SetVerbose(Verbose) - output.ClusterPrintData(&resp.Data.Rows[0]) - return nil -} +package clusters + +import ( + "fmt" + "strings" + "xcecli/common/output" + "xcecli/common/request" + "xcecli/config" + "xcecli/models" + + "github.com/spf13/cobra" +) + +var ( + cluster string + fields string + Verbose bool +) + +// NewClustersCmd create cluster query command +func NewClustersCmd(configManager *config.ConfigManager) *cobra.Command { + cmd := &cobra.Command{ + Use: "id", + GroupID: "query", + Short: "Query cluster info", + Long: `Query cluster info, support filter by cluster name . +example: + xce clusters # query all clusters`, + + RunE: func(cmd *cobra.Command, args []string) error { + err := runClusters(configManager) + if err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + //cmd.Flags().BoolVarP(&Verbose, "verbose", "l", false, "show cluster detail") + return cmd +} + +func runClusters(cm *config.ConfigManager) error { + // get the config + serverInfo, _ := cm.GetServerInfo() + + // create API client + apiClient := request.NewAPIClient(serverInfo.URL) + + // validate the parameters + cluster = strings.ToLower(cluster) + + // create the query parameters + params := make(map[string]string) + + // handle field selection + if fields != "" || cluster != "" { + params["fields"] = fields + } + + // create the API client and query the host info + var resp models.ClusterResponse + err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/xce/v1/clusters", &resp, params) + if err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to get clusters: [%s]", errMsg) + } + + if resp.Data.Count == 0 { + fmt.Println("no cluster found") + return nil + } + + // print result + output := output.NewOutput(fields, false, false) + output.SetVerbose(Verbose) + output.ClusterPrintData(&resp.Data.Rows[0]) + return nil +} diff --git a/src/frontend/xcecli/cmd/hosts/controller.go b/src/frontend/xcecli/cmd/hosts/controller.go index dcdc4ece..df5e6aa8 100755 --- a/src/frontend/xcecli/cmd/hosts/controller.go +++ b/src/frontend/xcecli/cmd/hosts/controller.go @@ -1,179 +1,177 @@ -package hosts - -import ( - "encoding/json" - "fmt" - "strings" - "xcecli/common/request" - "xcecli/config" - "xcecli/models" - - "github.com/spf13/cobra" -) - -const ( - HOST_OPEN = 1 - HOST_CLOSE = 2 -) - -// HostController host control controller -type HostController struct { - configManager *config.ConfigManager - action string - hostnames []string - comment string - allHosts bool -} - -// NewHostController create host control controller -func NewHostController(configManager *config.ConfigManager, action string) *HostController { - return &HostController{ - configManager: configManager, - action: action, - hostnames: []string{}, - } -} - -// Run execute host control -func (c *HostController) RunHostsControl() error { - // get server info and verify authentication - serverInfo, _ := c.configManager.GetServerInfo() - - // create API client - apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) - - // handle all hosts case - if c.allHosts { - c.hostnames = []string{"all"} - } - - // if no hostnames specified, return error - if len(c.hostnames) == 0 { - return fmt.Errorf("must specify hostnames or use --all option") - } - - // create request - req := models.HostControlRequest{ - ControlAction: c.action, - Hostname: c.hostnames, - Comment: c.comment, - } - - // encode request to JSON bytes - reqData, err := json.Marshal(req) - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to encode request data: [%s]", errMsg) - } - - var resp models.HostControlResponse - err = apiClient.PostRequestToAPIServer( - serverInfo.Token, - serverInfo.CACert, - "/xce/v1/hosts/control", - reqData, - &resp, - ) - - if err != nil { - return err - } - - if len(c.hostnames) == 1 && c.hostnames[0] == "all" { - for _, data := range resp.Data.Accepted { - fmt.Printf("host <%s> is being <%s>\n", data.HostName, c.action) - } - for _, host := range resp.Data.Failed { - fmt.Printf("host <%s> cannot be <%s> for reason <%s>\n", host.HostName, c.action, host.Msg) - } - } else { - for _, data := range resp.Data.Accepted { - fmt.Printf("host <%s> is being <%s>\n", data.HostName, c.action) - } - for _, host := range resp.Data.Failed { - fmt.Printf("host <%s> cannot be <%s> for reason <%s>\n", host.HostName, c.action, host.Msg) - } - } - return nil -} - -// Parse parse command line arguments -func (c *HostController) ParseCmdArgs(cmd *cobra.Command, args []string) error { - // handle hostnames - if len(args) > 0 { - c.hostnames = append(c.hostnames, args...) - } - - return nil -} - -// NewOpenHostsCmd create open hosts command -func NewOpenHostsCmd(configManager *config.ConfigManager) *cobra.Command { - - controller := NewHostController(configManager, "open") - - cmd := &cobra.Command{ - Use: "open [hostname1 hostname2...]", - Short: "Open hosts", - Long: `Open hosts by hostname. -Open hosts changes hosts from DOWN, UNREACH, or UNAVAIL status to OK status. - -Examples: - xce hosts open host1,host2,... # open specific hosts - xce hosts open --all # open all hosts`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunHostsControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.BoolVarP(&controller.allHosts, "all", "a", false, "apply to all hosts") - flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") - - return cmd -} - -// NewCloseHostsCmd create close hosts command -func NewCloseHostsCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewHostController(configManager, "close") - - cmd := &cobra.Command{ - Use: "close [hostname1 hostname2...]", - Short: "Close hosts", - Long: `Close hosts by hostname. -Close hosts changes hosts from OK, BUSY, or IDLE status to UNAVAIL status. - -Examples: - xce hosts close host1 host2 # close specific hosts - xce hosts close --all # close all hosts`, - RunE: func(cmd *cobra.Command, args []string) error { - - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunHostsControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.BoolVarP(&controller.allHosts, "all", "a", false, "apply to all hosts") - flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") - - return cmd -} +package hosts + +import ( + "encoding/json" + "fmt" + "strings" + "xcecli/common/request" + "xcecli/config" + "xcecli/models" + + "github.com/spf13/cobra" +) + +const ( + HOST_OPEN = 1 + HOST_CLOSE = 2 +) + +// HostController host control controller +type HostController struct { + configManager *config.ConfigManager + action string + hostnames []string + comment string + allHosts bool +} + +// NewHostController create host control controller +func NewHostController(configManager *config.ConfigManager, action string) *HostController { + return &HostController{ + configManager: configManager, + action: action, + hostnames: []string{}, + } +} + +// Run execute host control +func (c *HostController) RunHostsControl() error { + // get server info and verify authentication + serverInfo, _ := c.configManager.GetServerInfo() + + // create API client + apiClient := request.NewAPIClient(serverInfo.URL) + + // handle all hosts case + if c.allHosts { + c.hostnames = []string{"all"} + } + + // if no hostnames specified, return error + if len(c.hostnames) == 0 { + return fmt.Errorf("must specify hostnames or use --all option") + } + + // create request + req := models.HostControlRequest{ + ControlAction: c.action, + Hostname: c.hostnames, + Comment: c.comment, + } + + // encode request to JSON bytes + reqData, err := json.Marshal(req) + if err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to encode request data: [%s]", errMsg) + } + + var resp models.HostControlResponse + err = apiClient.PostRequestToAPIServer( + serverInfo.Token, + "/xce/v1/hosts/control", + reqData, + &resp, + ) + + if err != nil { + return err + } + + if len(c.hostnames) == 1 && c.hostnames[0] == "all" { + for _, data := range resp.Data.Accepted { + fmt.Printf("host <%s> is being <%s>\n", data.HostName, c.action) + } + for _, host := range resp.Data.Failed { + fmt.Printf("host <%s> cannot be <%s> for reason <%s>\n", host.HostName, c.action, host.Msg) + } + } else { + for _, data := range resp.Data.Accepted { + fmt.Printf("host <%s> is being <%s>\n", data.HostName, c.action) + } + for _, host := range resp.Data.Failed { + fmt.Printf("host <%s> cannot be <%s> for reason <%s>\n", host.HostName, c.action, host.Msg) + } + } + return nil +} + +// Parse parse command line arguments +func (c *HostController) ParseCmdArgs(cmd *cobra.Command, args []string) error { + // handle hostnames + if len(args) > 0 { + c.hostnames = append(c.hostnames, args...) + } + + return nil +} + +// NewOpenHostsCmd create open hosts command +func NewOpenHostsCmd(configManager *config.ConfigManager) *cobra.Command { + + controller := NewHostController(configManager, "open") + + cmd := &cobra.Command{ + Use: "open [hostname1 hostname2...]", + Short: "Open hosts", + Long: `Open hosts by hostname. +Open hosts changes hosts from DOWN, UNREACH, or UNAVAIL status to OK status. + +Examples: + xce hosts open host1,host2,... # open specific hosts + xce hosts open --all # open all hosts`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunHostsControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.BoolVarP(&controller.allHosts, "all", "a", false, "apply to all hosts") + flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") + + return cmd +} + +// NewCloseHostsCmd create close hosts command +func NewCloseHostsCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewHostController(configManager, "close") + + cmd := &cobra.Command{ + Use: "close [hostname1 hostname2...]", + Short: "Close hosts", + Long: `Close hosts by hostname. +Close hosts changes hosts from OK, BUSY, or IDLE status to UNAVAIL status. + +Examples: + xce hosts close host1 host2 # close specific hosts + xce hosts close --all # close all hosts`, + RunE: func(cmd *cobra.Command, args []string) error { + + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunHostsControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.BoolVarP(&controller.allHosts, "all", "a", false, "apply to all hosts") + flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") + + return cmd +} diff --git a/src/frontend/xcecli/cmd/hosts/list.go b/src/frontend/xcecli/cmd/hosts/list.go index 5d1b4d09..29b45038 100755 --- a/src/frontend/xcecli/cmd/hosts/list.go +++ b/src/frontend/xcecli/cmd/hosts/list.go @@ -1,199 +1,198 @@ -package hosts - -import ( - "fmt" - "strings" - "xcecli/common/output" - "xcecli/common/request" - "xcecli/config" - logger "xcecli/config" - - host_filter "xcecli/filters" - "xcecli/models" - - "go.uber.org/zap" - - "github.com/spf13/cobra" -) - -// HostsQuerier struct for querying hosts -type HostsQuerier struct { - configManager *config.ConfigManager - filter *host_filter.HostsFilter - fields string - jsonOutput bool - noHeader bool - verbose bool - wideFormat bool // wide format -} - -// NewHostsQuerier creates a new hosts querier -func NewHostsQuerier(configManager *config.ConfigManager) *HostsQuerier { - return &HostsQuerier{ - configManager: configManager, - filter: &host_filter.HostsFilter{}, - } -} - -// Parse parses command line arguments, simplified version that doesn't need to process flags already bound to filter -func (hq *HostsQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error { - // No additional processing needed, all parameters are bound in flags - if len(args) > 0 { - hq.filter.HostName = strings.Join(args, ",") - } - - // Define mutual exclusions for verbose flag - if hq.verbose { - conflicts := make([]string, 0) - - if hq.wideFormat { - conflicts = append(conflicts, "--wided (-w)") - } - - if hq.fields != "" { - conflicts = append(conflicts, "--output (-o)") - } - - if len(conflicts) > 0 { - return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", - strings.Join(conflicts, ", "), cmd.UsageString()) - } - } - return nil -} - -// Run executes the hosts query command -func (hq *HostsQuerier) RunHostsList() error { - // Get configuration - - // Get server information - serverInfo, _ := hq.configManager.GetServerInfo() - // create API client - apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) - - // Create query parameters - params := make(map[string]string) - - // Get filter string - filterStr, err := hq.filter.HandleHostFilter() - - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to get filter: [%s]", errMsg) - } else if filterStr != "" { - params["filter"] = filterStr - } - - // Field mapping - fieldsMap := models.HostFieldsMap - - if hq.verbose && hq.fields == "" { - for header := range fieldsMap { - hq.fields += header + "," - } - hq.fields = strings.TrimSuffix(hq.fields, ",") - } - // Process field selection - if hq.fields != "" { - hq.fields = request.CleanFields(hq.fields) - mappedFields, err := request.CheckAndMapFields(hq.fields, fieldsMap) - if err != nil { - return err - } - params["fields"] = strings.Join(mappedFields, ",") - } - - // create page printer - pagePrinter := output.NewPagePrinter(output.DataTypeHosts, output.WithFields(hq.fields), - output.WithFormatJson(hq.jsonOutput), output.WithNoHeader(hq.noHeader), - output.WithVerbose(hq.verbose), output.WithWideFormat(hq.wideFormat)) - - // Record if the first page has been processed - isFirstPage := true - - var ( - resp models.HostsResponse - loopCount int - pageToken string - ) - - // Loop to get all pages of data - for { - if loopCount > 0 && pageToken == "" { - break - } - - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/hosts", &resp, params) - if err != nil { - return err - } - - if len(resp.Data.Rows) == 0 { - if isFirstPage { - fmt.Println("no host found") - return nil - } - } - - // Process the first page - if isFirstPage { - // Start printing process - pagePrinter.StartPrinting() - // Prepare for printing, calculate column width and table header, etc. - pagePrinter.PrepareForPrinting(&resp) - isFirstPage = false - } - - // Print the current page - pagePrinter.PrintPage(&resp) - - // Update page token and loop counter - pageToken = resp.Data.Page_token - loopCount++ - } - - // Finish printing - pagePrinter.Finish() - - logger.Debug("hosts response completed", zap.Int("pages", loopCount)) - return nil -} - -func NewListHostsCmd(configManager *config.ConfigManager) *cobra.Command { - hostsQuerier := NewHostsQuerier(configManager) - - cmd := &cobra.Command{ - Use: "list", - Short: "Query the host info", - Long: `Query the host info, support basic info and detailed info. -Example: - xce hosts list # query the full info - xce hosts list -a # query the detailed info for all hosts - xce hosts list -o "host_name status max_cpus .." # query the specified fields - xce hosts list -o "host_name status max_cpus .." --json # query the specified fields and output in json format`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := hostsQuerier.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := hostsQuerier.RunHostsList(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // Add command line parameters - flags := cmd.Flags() - flags.BoolVarP(&hostsQuerier.verbose, "host_info", "a", false, "host information") - flags.StringVarP(&hostsQuerier.filter.HostType, "host_type", "t", "", "host type (X86_64 or ARM)") - flags.StringVarP(&hostsQuerier.fields, "output", "o", "", "output fields for hosts") - flags.BoolVarP(&hostsQuerier.wideFormat, "wided", "w", false, "output in wide format") - flags.BoolVarP(&hostsQuerier.verbose, "verbose", "l", false, "show host detail") - flags.BoolVarP(&hostsQuerier.jsonOutput, "json", "", false, "output in JSON format") - flags.BoolVarP(&hostsQuerier.noHeader, "noheader", "n", false, "output no header") - - return cmd -} +package hosts + +import ( + "fmt" + "strings" + "xcecli/common/output" + "xcecli/common/request" + "xcecli/config" + logger "xcecli/config" + + host_filter "xcecli/filters" + "xcecli/models" + + "go.uber.org/zap" + + "github.com/spf13/cobra" +) + +// HostsQuerier struct for querying hosts +type HostsQuerier struct { + configManager *config.ConfigManager + filter *host_filter.HostsFilter + fields string + jsonOutput bool + noHeader bool + verbose bool + wideFormat bool // wide format +} + +// NewHostsQuerier creates a new hosts querier +func NewHostsQuerier(configManager *config.ConfigManager) *HostsQuerier { + return &HostsQuerier{ + configManager: configManager, + filter: &host_filter.HostsFilter{}, + } +} + +// Parse parses command line arguments, simplified version that doesn't need to process flags already bound to filter +func (hq *HostsQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error { + // No additional processing needed, all parameters are bound in flags + if len(args) > 0 { + hq.filter.HostName = strings.Join(args, ",") + } + + // Define mutual exclusions for verbose flag + if hq.verbose { + conflicts := make([]string, 0) + + if hq.wideFormat { + conflicts = append(conflicts, "--wided (-w)") + } + + if hq.fields != "" { + conflicts = append(conflicts, "--output (-o)") + } + + if len(conflicts) > 0 { + return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", + strings.Join(conflicts, ", "), cmd.UsageString()) + } + } + return nil +} + +// Run executes the hosts query command +func (hq *HostsQuerier) RunHostsList() error { + // Get configuration + + // Get server information + serverInfo, _ := hq.configManager.GetServerInfo() + // create API client + apiClient := request.NewAPIClient(serverInfo.URL) + + // Create query parameters + params := make(map[string]string) + + // Get filter string + filterStr, err := hq.filter.HandleHostFilter() + + if err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to get filter: [%s]", errMsg) + } else if filterStr != "" { + params["filter"] = filterStr + } + + // Field mapping + fieldsMap := models.HostFieldsMap + + if hq.verbose && hq.fields == "" { + for header := range fieldsMap { + hq.fields += header + "," + } + hq.fields = strings.TrimSuffix(hq.fields, ",") + } + // Process field selection + if hq.fields != "" { + hq.fields = request.CleanFields(hq.fields) + mappedFields, err := request.CheckAndMapFields(hq.fields, fieldsMap) + if err != nil { + return err + } + params["fields"] = strings.Join(mappedFields, ",") + } + + // create page printer + pagePrinter := output.NewPagePrinter(output.DataTypeHosts, output.WithFields(hq.fields), + output.WithFormatJson(hq.jsonOutput), output.WithNoHeader(hq.noHeader), + output.WithVerbose(hq.verbose), output.WithWideFormat(hq.wideFormat)) + + // Record if the first page has been processed + isFirstPage := true + + var ( + resp models.HostsResponse + loopCount int + pageToken string + ) + + // Loop to get all pages of data + for { + if loopCount > 0 && pageToken == "" { + break + } + + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/hosts", &resp, params) + if err != nil { + return err + } + + if len(resp.Data.Rows) == 0 { + if isFirstPage { + fmt.Println("no host found") + return nil + } + } + + // Process the first page + if isFirstPage { + // Start printing process + pagePrinter.StartPrinting() + // Prepare for printing, calculate column width and table header, etc. + pagePrinter.PrepareForPrinting(&resp) + isFirstPage = false + } + + // Print the current page + pagePrinter.PrintPage(&resp) + + // Update page token and loop counter + pageToken = resp.Data.Page_token + loopCount++ + } + + // Finish printing + pagePrinter.Finish() + + logger.Debug("hosts response completed", zap.Int("pages", loopCount)) + return nil +} + +func NewListHostsCmd(configManager *config.ConfigManager) *cobra.Command { + hostsQuerier := NewHostsQuerier(configManager) + + cmd := &cobra.Command{ + Use: "list", + Short: "Query the host info", + Long: `Query the host info, support basic info and detailed info. +Example: + xce hosts list # query the full info + xce hosts list -a # query the detailed info for all hosts + xce hosts list -o "host_name status max_cpus .." # query the specified fields + xce hosts list -o "host_name status max_cpus .." --json # query the specified fields and output in json format`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := hostsQuerier.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := hostsQuerier.RunHostsList(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // Add command line parameters + flags := cmd.Flags() + flags.BoolVarP(&hostsQuerier.verbose, "host_info", "a", false, "host information") + flags.StringVarP(&hostsQuerier.filter.HostType, "host_type", "t", "", "host type (X86_64 or ARM)") + flags.StringVarP(&hostsQuerier.fields, "output", "o", "", "output fields for hosts") + flags.BoolVarP(&hostsQuerier.wideFormat, "wided", "w", false, "output in wide format") + flags.BoolVarP(&hostsQuerier.verbose, "verbose", "l", false, "show host detail") + flags.BoolVarP(&hostsQuerier.jsonOutput, "json", "", false, "output in JSON format") + flags.BoolVarP(&hostsQuerier.noHeader, "noheader", "n", false, "output no header") + + return cmd +} diff --git a/src/frontend/xcecli/cmd/hosts/select.go b/src/frontend/xcecli/cmd/hosts/select.go index 215ce72b..101aa642 100644 --- a/src/frontend/xcecli/cmd/hosts/select.go +++ b/src/frontend/xcecli/cmd/hosts/select.go @@ -36,7 +36,6 @@ func (s *HostSelector) run() error { return err } apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) req := models.ResReqSelectRequest{ Select: s.resReq, @@ -46,7 +45,7 @@ func (s *HostSelector) run() error { body, _ := json.Marshal(req) var resp models.HostsResponse - if err := apiClient.PostRequestToAPIServer(serverInfo.Token, serverInfo.CACert, + if err := apiClient.PostRequestToAPIServer(serverInfo.Token, "/xce/v1/hosts/select", body, &resp); err != nil { return err } diff --git a/src/frontend/xcecli/cmd/jobs/controller.go b/src/frontend/xcecli/cmd/jobs/controller.go index 7a7241de..866ced9b 100755 --- a/src/frontend/xcecli/cmd/jobs/controller.go +++ b/src/frontend/xcecli/cmd/jobs/controller.go @@ -1,287 +1,285 @@ -package jobs - -import ( - "encoding/json" - "fmt" - "strings" - "xcecli/common/format" - "xcecli/common/request" - "xcecli/config" - "xcecli/models" - - "github.com/spf13/cobra" -) - -// JobController defines job control -type JobController struct { - configManager *config.ConfigManager - action string - filter models.JobControlFilter - jobIDs []string -} - -// NewJobController create job controller -func NewJobController(configManager *config.ConfigManager, action string) *JobController { - return &JobController{ - configManager: configManager, - action: action, - } -} - -// Run execute job control -func (c *JobController) RunJobsControl() error { - // Get server information - serverInfo, _ := c.configManager.GetServerInfo() - - // Create API client - apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) - - if len(c.jobIDs) > 0 { - // if job IDs are specified, use job ID interface - return c.controlJobsByIDs(apiClient, serverInfo.Token, serverInfo.CACert) - } else { - // otherwise use filter conditions - return c.controlJobsByFilter(apiClient, serverInfo.Token, serverInfo.CACert) - } -} - -// control jobs by job IDs -func (c *JobController) controlJobsByIDs(apiClient *request.APIClient, token string, certPath string) error { - apiIDs := make([]string, 0, len(c.jobIDs)) - for _, id := range c.jobIDs { - apiID, err := format.ToAPIJobID(strings.TrimSpace(id)) - if err != nil { - return fmt.Errorf("invalid job ID %q: %w", id, err) - } - apiIDs = append(apiIDs, apiID) - } - url := fmt.Sprintf("/xce/v1/jobs/%s/control", strings.Join(apiIDs, ",")) - - req := models.JobControlJobIDRequest{ - Action: c.action, - } - // encode request data to JSON bytes - return c.printJobControlResponse(req, apiClient, token, certPath, url) -} - -// control jobs by filter conditions -func (c *JobController) controlJobsByFilter(apiClient *request.APIClient, token string, certPath string) error { - url := "/xce/v1/jobs/control" - - req := models.JobControlRequest{ - Action: c.action, - Filter: c.filter, - } - // encode request data to JSON bytes - return c.printJobControlResponse(req, apiClient, token, certPath, url) -} - -func (c *JobController) printJobControlResponse(req interface{}, apiClient *request.APIClient, token string, certPath string, url string) error { - reqData, err := json.Marshal(req) - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to encode request data: [%s]", errMsg) - } - var resp models.JobControlResponse - err = apiClient.PostRequestToAPIServer( - token, - certPath, - url, - reqData, - &resp, - ) - - if err != nil { - return err - } - for _, data := range resp.Data.Accepted { - fmt.Printf("job <%s> %s\n", format.FormatArrayChildID(data.JobID, data.ArrayIdx), strings.TrimPrefix(data.Msg, "job ")) - } - - for _, data := range resp.Data.Failed { - fmt.Printf("job <%s> %s\n", format.FormatArrayChildID(data.JobID, data.ArrayIdx), strings.TrimPrefix(data.Msg, "job ")) - } - return nil -} - -// Parse parse command line arguments -// Parse command line arguments -func (c *JobController) ParseCmdArgs(cmd *cobra.Command, args []string) error { - if len(args) > 0 { - if len(args) == 1 && args[0] == "0" { - c.filter.AllJobs = true - } else { - // parse job ID parameters - c.jobIDs = strings.Split(args[0], ",") - return nil - } - } - // check if any filter conditions are set - if c.filter.JobName == "" && c.filter.QueueName == "" && c.filter.Users == "" && !c.filter.AllJobs { - // if no job IDs or filter conditions are specified, default to processing all jobs of the current user - return fmt.Errorf("no job IDs or filter conditions specified") - } - return nil -} - -// NewKillJobsCmd create kill job command -func NewKillJobsCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewJobController(configManager, "kill") - - cmd := &cobra.Command{ - Use: "kill", - Short: "Kill jobs", - Long: `Kill jobs by job IDs or filters. -Examples: - xce jobs kill jobid1,jobid2,... # kill jobs by job IDs - xce jobs kill -J jobname # kill jobs by job name - xce jobs kill -q queue_name # kill jobs in a specific queue - xce jobs kill -u username # kill jobs by user`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunJobsControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") - flags.StringVarP(&controller.filter.QueueName, "queue", "q", "", "filter by queue name") - flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user") - flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") - - return cmd -} - -// NewStopJobsCmd create stop job command -func NewStopJobsCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewJobController(configManager, "stop") - - cmd := &cobra.Command{ - Use: "stop", - Short: "Stop jobs", - Long: `Stop jobs by job IDs or filters. -Examples: - xce jobs stop jobid1,jobid2,... # stop jobs by job IDs - xce jobs stop -J jobname # stop jobs by job name - xce jobs stop -q queue_name # stop jobs in a specific queue - xce jobs stop -u username # stop jobs by user`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunJobsControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") - flags.StringVarP(&controller.filter.QueueName, "queue", "q", "", "filter by queue name") - flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user") - flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") - - return cmd -} - -// NewResumeJobsCmd create resume job command -func NewResumeJobsCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewJobController(configManager, "resume") - - cmd := &cobra.Command{ - Use: "resume", - Short: "Resume jobs", - Long: `Resume jobs by job IDs or filters. -Examples: - xce jobs resume jobid1,jobid2,... # resume jobs by job IDs - xce jobs resume -J jobname # resume jobs by job name - xce jobs resume -u username # resume jobs by user`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunJobsControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") - flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user") - flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") - - return cmd -} - -func NewCheckPointJobsCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewJobController(configManager, "checkpoint") - - cmd := &cobra.Command{ - Use: "checkpoint", - Short: "Checkpoint jobs", - Long: `Checkpoint jobs by job IDs or filters. -Examples: - xce jobs checkpoint jobid1,jobid2,... # checkpoint jobs by job IDs`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunJobsControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - return cmd -} - -func NewRestartJobsCmd(configManager *config.ConfigManager) *cobra.Command { - restarter := NewJobRestart(configManager) - - cmd := &cobra.Command{ - Use: "restart", - Short: "Restart jobs", - Long: `Restart jobs by job ID. -Examples: - xce jobs restart jobid # restart job by jobid - xce jobs restart -m hostname # restart jobs by hostname`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := restarter.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := restarter.RunJobsRestart(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.StringVarP(&restarter.askedHosts, "hostname", "m", "", "restart on the specific host name") - return cmd -} +package jobs + +import ( + "encoding/json" + "fmt" + "strings" + "xcecli/common/format" + "xcecli/common/request" + "xcecli/config" + "xcecli/models" + + "github.com/spf13/cobra" +) + +// JobController defines job control +type JobController struct { + configManager *config.ConfigManager + action string + filter models.JobControlFilter + jobIDs []string +} + +// NewJobController create job controller +func NewJobController(configManager *config.ConfigManager, action string) *JobController { + return &JobController{ + configManager: configManager, + action: action, + } +} + +// Run execute job control +func (c *JobController) RunJobsControl() error { + // Get server information + serverInfo, _ := c.configManager.GetServerInfo() + + // Create API client + apiClient := request.NewAPIClient(serverInfo.URL) + + if len(c.jobIDs) > 0 { + // if job IDs are specified, use job ID interface + return c.controlJobsByIDs(apiClient, serverInfo.Token) + } else { + // otherwise use filter conditions + return c.controlJobsByFilter(apiClient, serverInfo.Token) + } +} + +// control jobs by job IDs +func (c *JobController) controlJobsByIDs(apiClient *request.APIClient, token string) error { + apiIDs := make([]string, 0, len(c.jobIDs)) + for _, id := range c.jobIDs { + apiID, err := format.ToAPIJobID(strings.TrimSpace(id)) + if err != nil { + return fmt.Errorf("invalid job ID %q: %w", id, err) + } + apiIDs = append(apiIDs, apiID) + } + url := fmt.Sprintf("/xce/v1/jobs/%s/control", strings.Join(apiIDs, ",")) + + req := models.JobControlJobIDRequest{ + Action: c.action, + } + // encode request data to JSON bytes + return c.printJobControlResponse(req, apiClient, token, url) +} + +// control jobs by filter conditions +func (c *JobController) controlJobsByFilter(apiClient *request.APIClient, token string) error { + url := "/xce/v1/jobs/control" + + req := models.JobControlRequest{ + Action: c.action, + Filter: c.filter, + } + // encode request data to JSON bytes + return c.printJobControlResponse(req, apiClient, token, url) +} + +func (c *JobController) printJobControlResponse(req interface{}, apiClient *request.APIClient, token string, url string) error { + reqData, err := json.Marshal(req) + if err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to encode request data: [%s]", errMsg) + } + var resp models.JobControlResponse + err = apiClient.PostRequestToAPIServer( + token, + url, + reqData, + &resp, + ) + + if err != nil { + return err + } + for _, data := range resp.Data.Accepted { + fmt.Printf("job <%s> %s\n", format.FormatArrayChildID(data.JobID, data.ArrayIdx), strings.TrimPrefix(data.Msg, "job ")) + } + + for _, data := range resp.Data.Failed { + fmt.Printf("job <%s> %s\n", format.FormatArrayChildID(data.JobID, data.ArrayIdx), strings.TrimPrefix(data.Msg, "job ")) + } + return nil +} + +// Parse parse command line arguments +// Parse command line arguments +func (c *JobController) ParseCmdArgs(cmd *cobra.Command, args []string) error { + if len(args) > 0 { + if len(args) == 1 && args[0] == "0" { + c.filter.AllJobs = true + } else { + // parse job ID parameters + c.jobIDs = strings.Split(args[0], ",") + return nil + } + } + // check if any filter conditions are set + if c.filter.JobName == "" && c.filter.QueueName == "" && c.filter.Users == "" && !c.filter.AllJobs { + // if no job IDs or filter conditions are specified, default to processing all jobs of the current user + return fmt.Errorf("no job IDs or filter conditions specified") + } + return nil +} + +// NewKillJobsCmd create kill job command +func NewKillJobsCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewJobController(configManager, "kill") + + cmd := &cobra.Command{ + Use: "kill", + Short: "Kill jobs", + Long: `Kill jobs by job IDs or filters. +Examples: + xce jobs kill jobid1,jobid2,... # kill jobs by job IDs + xce jobs kill -J jobname # kill jobs by job name + xce jobs kill -q queue_name # kill jobs in a specific queue + xce jobs kill -u username # kill jobs by user`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunJobsControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") + flags.StringVarP(&controller.filter.QueueName, "queue", "q", "", "filter by queue name") + flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user") + flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") + + return cmd +} + +// NewStopJobsCmd create stop job command +func NewStopJobsCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewJobController(configManager, "stop") + + cmd := &cobra.Command{ + Use: "stop", + Short: "Stop jobs", + Long: `Stop jobs by job IDs or filters. +Examples: + xce jobs stop jobid1,jobid2,... # stop jobs by job IDs + xce jobs stop -J jobname # stop jobs by job name + xce jobs stop -q queue_name # stop jobs in a specific queue + xce jobs stop -u username # stop jobs by user`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunJobsControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") + flags.StringVarP(&controller.filter.QueueName, "queue", "q", "", "filter by queue name") + flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user") + flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") + + return cmd +} + +// NewResumeJobsCmd create resume job command +func NewResumeJobsCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewJobController(configManager, "resume") + + cmd := &cobra.Command{ + Use: "resume", + Short: "Resume jobs", + Long: `Resume jobs by job IDs or filters. +Examples: + xce jobs resume jobid1,jobid2,... # resume jobs by job IDs + xce jobs resume -J jobname # resume jobs by job name + xce jobs resume -u username # resume jobs by user`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunJobsControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") + flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user") + flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") + + return cmd +} + +func NewCheckPointJobsCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewJobController(configManager, "checkpoint") + + cmd := &cobra.Command{ + Use: "checkpoint", + Short: "Checkpoint jobs", + Long: `Checkpoint jobs by job IDs or filters. +Examples: + xce jobs checkpoint jobid1,jobid2,... # checkpoint jobs by job IDs`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunJobsControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + return cmd +} + +func NewRestartJobsCmd(configManager *config.ConfigManager) *cobra.Command { + restarter := NewJobRestart(configManager) + + cmd := &cobra.Command{ + Use: "restart", + Short: "Restart jobs", + Long: `Restart jobs by job ID. +Examples: + xce jobs restart jobid # restart job by jobid + xce jobs restart -m hostname # restart jobs by hostname`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := restarter.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := restarter.RunJobsRestart(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.StringVarP(&restarter.askedHosts, "hostname", "m", "", "restart on the specific host name") + return cmd +} diff --git a/src/frontend/xcecli/cmd/jobs/list.go b/src/frontend/xcecli/cmd/jobs/list.go index 36686ddc..1db79290 100755 --- a/src/frontend/xcecli/cmd/jobs/list.go +++ b/src/frontend/xcecli/cmd/jobs/list.go @@ -1,369 +1,368 @@ -package jobs - -import ( - "fmt" - "strings" - "xcecli/common/output" - "xcecli/common/request" - "xcecli/config" - job_filter "xcecli/filters" - "xcecli/models" - - "github.com/spf13/cobra" -) - -var ( - resp models.JobsResponse - loopCount int - pageToken string - arrayJobsStatsResp models.ArrayJobsStatsResponse -) - -// JobsQuerier struct for querying jobs -type JobsQuerier struct { - configManager *config.ConfigManager - filter *job_filter.JobsFilter - fields string - jsonOutput bool - noHeader bool - verbose bool - wideFormat bool // wide format - arrayFlag bool // show array job statistics - arrayElementsFlag bool // show array job elements (sub-tasks of array jobs) - -} - -// NewJobsQuerier creates a new jobs querier -func NewJobsQuerier(configManager *config.ConfigManager) *JobsQuerier { - return &JobsQuerier{ - configManager: configManager, - filter: &job_filter.JobsFilter{}, - } - -} - -// Parses command line arguments, simplified version that only processes arguments not directly bound to filter -func (jq *JobsQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error { - // Process JobIDs parameter, which needs special handling - // input multiple jobids, it will use , to connect - if len(args) > 0 { - jq.filter.JobIDs = strings.Join(args, ",") - } - - // Define mutual exclusions for verbose flag - if jq.verbose { - conflicts := make([]string, 0) - - if jq.wideFormat { - conflicts = append(conflicts, "--wided (-w)") - } - - if jq.fields != "" { - conflicts = append(conflicts, "--output (-o)") - } - - if len(conflicts) > 0 { - return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", - strings.Join(conflicts, ", "), cmd.UsageString()) - } - } - return nil -} - -// Run executes the jobs query command -func (jq *JobsQuerier) RunJobsList() error { - // Get server information - serverInfo, _ := jq.configManager.GetServerInfo() - // create API client - apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) - - // if array job statistics - if jq.arrayFlag { - if jq.verbose { - fmt.Println("not support -A and -l parameters together") - return nil - } - return jq.runArrayJobStats(serverInfo, apiClient) - } - - // if array job elements (sub-tasks of array jobs) - if jq.arrayElementsFlag { - return jq.runArrayJobElements(serverInfo, apiClient) - } - - // the following is the regular job query logic - // Build query parameters - params := make(map[string]string) - //fmt.Println(params) - - // Get filter string - filterStr, err := jq.filter.HandleJobFilter("") - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to get parameter: [%s]", errMsg) - } else if filterStr != "" { - params["filter"] = filterStr - } - - // Field mapping - fieldsMap := models.JobFieldsMap - if jq.verbose && jq.fields == "" { - for header := range fieldsMap { - jq.fields += header + "," - } - jq.fields = strings.TrimSuffix(jq.fields, ",") - } - - // Process field selection - if jq.fields != "" { - jq.fields = request.CleanFields(jq.fields) - mappedFields, err := request.CheckAndMapFields(jq.fields, fieldsMap) - if err != nil { - return err - } - params["fields"] = strings.Join(mappedFields, ",") - } - - // create page printer - pagePrinter := output.NewPagePrinter(output.DataTypeJobs, output.WithFields(jq.fields), - output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), - output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat)) - - // record if the first page has been processed - isFirstPage := true - - // Loop to get all pages of data - loopCount = 0 - pageToken = "" - - for { - if loopCount > 0 && pageToken == "" { - break - } - - // Only add page_token to params if it's not empty - //if pageToken != "" { - //params["page_token"] = pageToken - //} - - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/jobs", &resp, params) - if err != nil { - return err - } - if len(resp.Data.Rows) == 0 { - if isFirstPage { - fmt.Println("no job found") - return nil - } - } - // process the first page - if isFirstPage { - // start printing process - pagePrinter.StartPrinting() - // - pagePrinter.PrepareForPrinting(&resp) - isFirstPage = false - } - - // print current page - pagePrinter.PrintPage(&resp) - - // update page token and loop count - pageToken = resp.Data.Page_token - - loopCount++ - } - - // finish printing - pagePrinter.Finish() - - return nil -} - -// runArrayJobStats process array job statistics -func (jq *JobsQuerier) runArrayJobStats(serverInfo *config.APIServerInfo, apiClient *request.APIClient) error { - // set query parameters, request array job statistics - params := map[string]string{ - "type": "array_stats", - } - params["type"] = "array_stats" - - filterStr, err := jq.filter.HandleJobFilter("array_stats") - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to get parameter: [%s]", errMsg) - } - - if filterStr != "" { - params["filter"] = filterStr - } - - // create response object - - // create array job page printer - pagePrinter := output.NewPagePrinter(output.DataTypeArrayJobs, output.WithFields(jq.fields), - output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), - output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat)) - - // record if the first page has been processed - isFirstPage := true - - // Loop to get all pages of data - loopCount = 0 - pageToken = "" - - for { - if loopCount > 0 && pageToken == "" { - break - } - - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/jobs", &arrayJobsStatsResp, params) - if err != nil { - return err - } - if len(arrayJobsStatsResp.Data.Rows) == 0 { - if isFirstPage { - fmt.Println("no job found") - return nil - } - } - // process the first page - if isFirstPage { - // start printing process - pagePrinter.StartPrinting() - // - pagePrinter.PrepareForPrinting(&arrayJobsStatsResp) - isFirstPage = false - } - - // print current page - pagePrinter.PrintPage(&arrayJobsStatsResp) - - // update page token and loop count - pageToken = arrayJobsStatsResp.Data.Page_token - - loopCount++ - } - // finish printing - pagePrinter.Finish() - - return nil -} - -// runArrayJobElements process array job elements -func (jq *JobsQuerier) runArrayJobElements(serverInfo *config.APIServerInfo, apiClient *request.APIClient) error { - // set query parameters, request array job elements - params := map[string]string{ - "filter": "array_idx:ne:0", // query array_idx not equal to 0, that is, the sub-tasks of array jobs - } - - // create response object - var jobsResp models.JobsResponse - - // request array job elements data - err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, "/xce/v1/jobs?limit=1000", &jobsResp, params) - if err != nil { - return err - } - - // create array job page printer - pagePrinter := output.NewPagePrinter(output.DataTypeArrayElements, output.WithFields(jq.fields), - output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), - output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat)) - // start printing process - pagePrinter.StartPrinting() - - // prepare and print data - pagePrinter.PrepareForPrinting(&jobsResp) - pagePrinter.PrintPage(&jobsResp) - - // finish printing - pagePrinter.Finish() - - return nil -} - -func NewListJobsCmd(configManager *config.ConfigManager) *cobra.Command { - jobsQuerier := NewJobsQuerier(configManager) - - // Create the list subcommand - listCmd := &cobra.Command{ - Use: "list", - Short: "Query jobs info", - Long: `Query jobs detail info. -Example: - xce jobs list # list all jobs - xce jobs list jobid0,jobid1,...jobidn # list jobs by job id - xce jobs list -a # list run,done,pending,suspended jobs - xce jobs list -d # list done jobs - xce jobs list -d -r # list done and running jobs - xce jobs list -d -r -s # list done and running and suspended jobs - xce jobs list -d -r -s -p # list done and running and suspended and pending jobs - xce jobs list -d -u all # list done jobs by all users - xce jobs list -d -r -s -p -u username # list done and running and suspended and pending jobs by user - xce jobs list -d -r -s -p -u username -q queuename # list done and running and suspended and pending jobs by user and queue - xce jobs list -j jobname # list jobs by job name - xce jobs list -Jd jobdescription # list jobs by job description - xce jobs list --Lp licenseprojectname # list license project jobs - xce jobs list -o "jobid submit_user ..." # list jobs by fields - xce jobs list -o "jobid submit_user ..." --json # list jobs by fields and output in json format - xce jobs list -p # list pending jobs - xce jobs list -p -u username # list pending jobs by user - xce jobs list -p -u all # list pending jobs by all users - xce jobs list -P projectname # show project jobs - xce jobs list -q queuename # list jobs by queue - xce jobs list -q queuename -u username # list jobs by user in queue - xce jobs list -r # list running jobs - xce jobs list -r -u username # list running jobs by user - xce jobs list -s # list suspended jobs - xce jobs list -s -u username # list suspended jobs by user - xce jobs list -s -u all # list suspended jobs by all users - xce jobs list --sla serviceclassname # list jobs by service class - xce jobs list -u all # list jobs by all users - xce jobs list -u username # list jobs by user - xce jobs list -u username all # list jobs by all user - xce jobs list -u username -q queuename # list jobs by user and queue - xce jobs list -u all -q queuename # list jobs by all users and queue - xce jobs list -q -u all # list jobs by all users and all queues - xce jobs list -A # list array job statistics - xce jobs list -W # list array job elements - xce jobs list -m host1 # list jobs executing on specific host - xce jobs list -m host1,host2 # list jobs executing on multiple hosts`, - - RunE: func(cmd *cobra.Command, args []string) error { - if err := jobsQuerier.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := jobsQuerier.RunJobsList(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - // Add command line parameters - flags := listCmd.Flags() - flags.BoolVarP(&jobsQuerier.filter.AllJobInfoFlag, "job", "a", false, "show all jobs status") - flags.StringVarP(&jobsQuerier.filter.User, "user", "u", "", "filter by user") - flags.StringVarP(&jobsQuerier.filter.Queue, "queue", "q", "", "filter by queue") - flags.BoolVarP(&jobsQuerier.filter.PendingFlag, "pending", "p", false, "show pending jobs") - flags.StringVarP(&jobsQuerier.filter.ProjectName, "project", "P", "", "show project jobs") - flags.BoolVarP(&jobsQuerier.filter.RunningFlag, "running", "r", false, "show running jobs") - flags.BoolVarP(&jobsQuerier.filter.DoneFlag, "done", "d", false, "show done jobs") - flags.BoolVarP(&jobsQuerier.filter.SSUSPFlag, "ssuspFlag", "s", false, "show suspended jobs") - flags.StringVarP(&jobsQuerier.filter.JobName, "job_name", "J", "", "filter by job name") - flags.StringVarP(&jobsQuerier.filter.LicenseProject, "Lp", "", "", "show license project jobs") - flags.StringVarP(&jobsQuerier.filter.JobDescription, "Jd", "", "", "show job description") - flags.StringVarP(&jobsQuerier.filter.ServiceClass, "sla", "", "", "show job by service class") - flags.BoolVarP(&jobsQuerier.arrayFlag, "array", "A", false, "show array job statistics") - //flags.BoolVarP(&jobsQuerier.arrayElementsFlag, "wide", "w", false, "show array job elements (sub-tasks of array jobs)") - flags.StringVarP(&jobsQuerier.filter.ExecHosts, "exechost", "m", "", "filter jobs by execution host, comma separated for multiple hosts") - flags.StringVarP(&jobsQuerier.fields, "output", "o", "", "output fields for jobs") - flags.BoolVarP(&jobsQuerier.wideFormat, "wided", "w", false, "output in wide format") - flags.BoolVarP(&jobsQuerier.verbose, "verbose", "l", false, "show job detail") - flags.BoolVarP(&jobsQuerier.jsonOutput, "json", "", false, "output in JSON format") - flags.BoolVarP(&jobsQuerier.noHeader, "noheader", "n", false, "output no header") - return listCmd -} +package jobs + +import ( + "fmt" + "strings" + "xcecli/common/output" + "xcecli/common/request" + "xcecli/config" + job_filter "xcecli/filters" + "xcecli/models" + + "github.com/spf13/cobra" +) + +var ( + resp models.JobsResponse + loopCount int + pageToken string + arrayJobsStatsResp models.ArrayJobsStatsResponse +) + +// JobsQuerier struct for querying jobs +type JobsQuerier struct { + configManager *config.ConfigManager + filter *job_filter.JobsFilter + fields string + jsonOutput bool + noHeader bool + verbose bool + wideFormat bool // wide format + arrayFlag bool // show array job statistics + arrayElementsFlag bool // show array job elements (sub-tasks of array jobs) + +} + +// NewJobsQuerier creates a new jobs querier +func NewJobsQuerier(configManager *config.ConfigManager) *JobsQuerier { + return &JobsQuerier{ + configManager: configManager, + filter: &job_filter.JobsFilter{}, + } + +} + +// Parses command line arguments, simplified version that only processes arguments not directly bound to filter +func (jq *JobsQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error { + // Process JobIDs parameter, which needs special handling + // input multiple jobids, it will use , to connect + if len(args) > 0 { + jq.filter.JobIDs = strings.Join(args, ",") + } + + // Define mutual exclusions for verbose flag + if jq.verbose { + conflicts := make([]string, 0) + + if jq.wideFormat { + conflicts = append(conflicts, "--wided (-w)") + } + + if jq.fields != "" { + conflicts = append(conflicts, "--output (-o)") + } + + if len(conflicts) > 0 { + return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", + strings.Join(conflicts, ", "), cmd.UsageString()) + } + } + return nil +} + +// Run executes the jobs query command +func (jq *JobsQuerier) RunJobsList() error { + // Get server information + serverInfo, _ := jq.configManager.GetServerInfo() + // create API client + apiClient := request.NewAPIClient(serverInfo.URL) + + // if array job statistics + if jq.arrayFlag { + if jq.verbose { + fmt.Println("not support -A and -l parameters together") + return nil + } + return jq.runArrayJobStats(serverInfo, apiClient) + } + + // if array job elements (sub-tasks of array jobs) + if jq.arrayElementsFlag { + return jq.runArrayJobElements(serverInfo, apiClient) + } + + // the following is the regular job query logic + // Build query parameters + params := make(map[string]string) + //fmt.Println(params) + + // Get filter string + filterStr, err := jq.filter.HandleJobFilter("") + if err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to get parameter: [%s]", errMsg) + } else if filterStr != "" { + params["filter"] = filterStr + } + + // Field mapping + fieldsMap := models.JobFieldsMap + if jq.verbose && jq.fields == "" { + for header := range fieldsMap { + jq.fields += header + "," + } + jq.fields = strings.TrimSuffix(jq.fields, ",") + } + + // Process field selection + if jq.fields != "" { + jq.fields = request.CleanFields(jq.fields) + mappedFields, err := request.CheckAndMapFields(jq.fields, fieldsMap) + if err != nil { + return err + } + params["fields"] = strings.Join(mappedFields, ",") + } + + // create page printer + pagePrinter := output.NewPagePrinter(output.DataTypeJobs, output.WithFields(jq.fields), + output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), + output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat)) + + // record if the first page has been processed + isFirstPage := true + + // Loop to get all pages of data + loopCount = 0 + pageToken = "" + + for { + if loopCount > 0 && pageToken == "" { + break + } + + // Only add page_token to params if it's not empty + //if pageToken != "" { + //params["page_token"] = pageToken + //} + + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/jobs", &resp, params) + if err != nil { + return err + } + if len(resp.Data.Rows) == 0 { + if isFirstPage { + fmt.Println("no job found") + return nil + } + } + // process the first page + if isFirstPage { + // start printing process + pagePrinter.StartPrinting() + // + pagePrinter.PrepareForPrinting(&resp) + isFirstPage = false + } + + // print current page + pagePrinter.PrintPage(&resp) + + // update page token and loop count + pageToken = resp.Data.Page_token + + loopCount++ + } + + // finish printing + pagePrinter.Finish() + + return nil +} + +// runArrayJobStats process array job statistics +func (jq *JobsQuerier) runArrayJobStats(serverInfo *config.APIServerInfo, apiClient *request.APIClient) error { + // set query parameters, request array job statistics + params := map[string]string{ + "type": "array_stats", + } + params["type"] = "array_stats" + + filterStr, err := jq.filter.HandleJobFilter("array_stats") + if err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to get parameter: [%s]", errMsg) + } + + if filterStr != "" { + params["filter"] = filterStr + } + + // create response object + + // create array job page printer + pagePrinter := output.NewPagePrinter(output.DataTypeArrayJobs, output.WithFields(jq.fields), + output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), + output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat)) + + // record if the first page has been processed + isFirstPage := true + + // Loop to get all pages of data + loopCount = 0 + pageToken = "" + + for { + if loopCount > 0 && pageToken == "" { + break + } + + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/jobs", &arrayJobsStatsResp, params) + if err != nil { + return err + } + if len(arrayJobsStatsResp.Data.Rows) == 0 { + if isFirstPage { + fmt.Println("no job found") + return nil + } + } + // process the first page + if isFirstPage { + // start printing process + pagePrinter.StartPrinting() + // + pagePrinter.PrepareForPrinting(&arrayJobsStatsResp) + isFirstPage = false + } + + // print current page + pagePrinter.PrintPage(&arrayJobsStatsResp) + + // update page token and loop count + pageToken = arrayJobsStatsResp.Data.Page_token + + loopCount++ + } + // finish printing + pagePrinter.Finish() + + return nil +} + +// runArrayJobElements process array job elements +func (jq *JobsQuerier) runArrayJobElements(serverInfo *config.APIServerInfo, apiClient *request.APIClient) error { + // set query parameters, request array job elements + params := map[string]string{ + "filter": "array_idx:ne:0", // query array_idx not equal to 0, that is, the sub-tasks of array jobs + } + + // create response object + var jobsResp models.JobsResponse + + // request array job elements data + err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/xce/v1/jobs?limit=1000", &jobsResp, params) + if err != nil { + return err + } + + // create array job page printer + pagePrinter := output.NewPagePrinter(output.DataTypeArrayElements, output.WithFields(jq.fields), + output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), + output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat)) + // start printing process + pagePrinter.StartPrinting() + + // prepare and print data + pagePrinter.PrepareForPrinting(&jobsResp) + pagePrinter.PrintPage(&jobsResp) + + // finish printing + pagePrinter.Finish() + + return nil +} + +func NewListJobsCmd(configManager *config.ConfigManager) *cobra.Command { + jobsQuerier := NewJobsQuerier(configManager) + + // Create the list subcommand + listCmd := &cobra.Command{ + Use: "list", + Short: "Query jobs info", + Long: `Query jobs detail info. +Example: + xce jobs list # list all jobs + xce jobs list jobid0,jobid1,...jobidn # list jobs by job id + xce jobs list -a # list run,done,pending,suspended jobs + xce jobs list -d # list done jobs + xce jobs list -d -r # list done and running jobs + xce jobs list -d -r -s # list done and running and suspended jobs + xce jobs list -d -r -s -p # list done and running and suspended and pending jobs + xce jobs list -d -u all # list done jobs by all users + xce jobs list -d -r -s -p -u username # list done and running and suspended and pending jobs by user + xce jobs list -d -r -s -p -u username -q queuename # list done and running and suspended and pending jobs by user and queue + xce jobs list -j jobname # list jobs by job name + xce jobs list -Jd jobdescription # list jobs by job description + xce jobs list --Lp licenseprojectname # list license project jobs + xce jobs list -o "jobid submit_user ..." # list jobs by fields + xce jobs list -o "jobid submit_user ..." --json # list jobs by fields and output in json format + xce jobs list -p # list pending jobs + xce jobs list -p -u username # list pending jobs by user + xce jobs list -p -u all # list pending jobs by all users + xce jobs list -P projectname # show project jobs + xce jobs list -q queuename # list jobs by queue + xce jobs list -q queuename -u username # list jobs by user in queue + xce jobs list -r # list running jobs + xce jobs list -r -u username # list running jobs by user + xce jobs list -s # list suspended jobs + xce jobs list -s -u username # list suspended jobs by user + xce jobs list -s -u all # list suspended jobs by all users + xce jobs list --sla serviceclassname # list jobs by service class + xce jobs list -u all # list jobs by all users + xce jobs list -u username # list jobs by user + xce jobs list -u username all # list jobs by all user + xce jobs list -u username -q queuename # list jobs by user and queue + xce jobs list -u all -q queuename # list jobs by all users and queue + xce jobs list -q -u all # list jobs by all users and all queues + xce jobs list -A # list array job statistics + xce jobs list -W # list array job elements + xce jobs list -m host1 # list jobs executing on specific host + xce jobs list -m host1,host2 # list jobs executing on multiple hosts`, + + RunE: func(cmd *cobra.Command, args []string) error { + if err := jobsQuerier.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := jobsQuerier.RunJobsList(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + // Add command line parameters + flags := listCmd.Flags() + flags.BoolVarP(&jobsQuerier.filter.AllJobInfoFlag, "job", "a", false, "show all jobs status") + flags.StringVarP(&jobsQuerier.filter.User, "user", "u", "", "filter by user") + flags.StringVarP(&jobsQuerier.filter.Queue, "queue", "q", "", "filter by queue") + flags.BoolVarP(&jobsQuerier.filter.PendingFlag, "pending", "p", false, "show pending jobs") + flags.StringVarP(&jobsQuerier.filter.ProjectName, "project", "P", "", "show project jobs") + flags.BoolVarP(&jobsQuerier.filter.RunningFlag, "running", "r", false, "show running jobs") + flags.BoolVarP(&jobsQuerier.filter.DoneFlag, "done", "d", false, "show done jobs") + flags.BoolVarP(&jobsQuerier.filter.SSUSPFlag, "ssuspFlag", "s", false, "show suspended jobs") + flags.StringVarP(&jobsQuerier.filter.JobName, "job_name", "J", "", "filter by job name") + flags.StringVarP(&jobsQuerier.filter.LicenseProject, "Lp", "", "", "show license project jobs") + flags.StringVarP(&jobsQuerier.filter.JobDescription, "Jd", "", "", "show job description") + flags.StringVarP(&jobsQuerier.filter.ServiceClass, "sla", "", "", "show job by service class") + flags.BoolVarP(&jobsQuerier.arrayFlag, "array", "A", false, "show array job statistics") + //flags.BoolVarP(&jobsQuerier.arrayElementsFlag, "wide", "w", false, "show array job elements (sub-tasks of array jobs)") + flags.StringVarP(&jobsQuerier.filter.ExecHosts, "exechost", "m", "", "filter jobs by execution host, comma separated for multiple hosts") + flags.StringVarP(&jobsQuerier.fields, "output", "o", "", "output fields for jobs") + flags.BoolVarP(&jobsQuerier.wideFormat, "wided", "w", false, "output in wide format") + flags.BoolVarP(&jobsQuerier.verbose, "verbose", "l", false, "show job detail") + flags.BoolVarP(&jobsQuerier.jsonOutput, "json", "", false, "output in JSON format") + flags.BoolVarP(&jobsQuerier.noHeader, "noheader", "n", false, "output no header") + return listCmd +} diff --git a/src/frontend/xcecli/cmd/jobs/restart.go b/src/frontend/xcecli/cmd/jobs/restart.go index f2bc5566..171c299a 100644 --- a/src/frontend/xcecli/cmd/jobs/restart.go +++ b/src/frontend/xcecli/cmd/jobs/restart.go @@ -44,23 +44,22 @@ func (c *JobRestart) RunJobsRestart() error { // Create API client apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) - return c.restartJobsByID(apiClient, serverInfo.Token, serverInfo.CACert) + return c.restartJobsByID(apiClient, serverInfo.Token) } -func (c *JobRestart) restartJobsByID(apiClient *request.APIClient, token string, certPath string) error { +func (c *JobRestart) restartJobsByID(apiClient *request.APIClient, token string) error { url := fmt.Sprintf("/xce/v1/jobs/%s/restart", c.jobID) req := &models.JobRestartRequest{ AskedHosts: c.askedHosts, } - return c.printJobRestartResponse(req, apiClient, token, certPath, url) + return c.printJobRestartResponse(req, apiClient, token, url) } -func (c *JobRestart) printJobRestartResponse(req interface{}, apiClient *request.APIClient, token string, certPath string, url string) error { +func (c *JobRestart) printJobRestartResponse(req interface{}, apiClient *request.APIClient, token string, url string) error { reqData, err := json.Marshal(req) if err != nil { errMsg := request.ExtractErrorMessage(err) @@ -69,7 +68,6 @@ func (c *JobRestart) printJobRestartResponse(req interface{}, apiClient *request var resp models.JobRestartResponse err = apiClient.PostRequestToAPIServer( token, - certPath, url, reqData, &resp, diff --git a/src/frontend/xcecli/cmd/jobs/submit.go b/src/frontend/xcecli/cmd/jobs/submit.go index 789a0fb3..8ccd2c29 100755 --- a/src/frontend/xcecli/cmd/jobs/submit.go +++ b/src/frontend/xcecli/cmd/jobs/submit.go @@ -462,7 +462,6 @@ func (js *JobSubmitter) RunJobsSubmit() error { // create the API client and submit the job apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) var transferFile []string if len(js.uploadFiles) > 0 { @@ -517,7 +516,6 @@ func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token strin err := apiClient.GetRequestFromAPIServer( "Bearer "+token, "", - "", fmt.Sprintf("/xce/v1/jobs/%s", jobID), &resp, map[string]string{ /*"fields": "interactive_job_url"*/ }, @@ -548,7 +546,6 @@ func (js *JobSubmitter) checkExtendData(apiClient *request.APIClient, token stri err := apiClient.GetRequestFromAPIServer( "Bearer "+token, "", - "", fmt.Sprintf("/xce/v1/jobs/%s", jobID), &resp, map[string]string{"fields": "interactive_job_url"}, diff --git a/src/frontend/xcecli/cmd/permissions/list.go b/src/frontend/xcecli/cmd/permissions/list.go index 5b73f2ae..04ee1ea3 100644 --- a/src/frontend/xcecli/cmd/permissions/list.go +++ b/src/frontend/xcecli/cmd/permissions/list.go @@ -76,14 +76,13 @@ func (pq *PermsQuerier) RunPermsList() error { // create API client apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // Build query parameters params := make(map[string]string) pq.fields = []string{} // Make API request - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, "/xce/v1/permissions", &resp, params) + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/xce/v1/permissions", &resp, params) if err != nil { return err } diff --git a/src/frontend/xcecli/cmd/queues/controller.go b/src/frontend/xcecli/cmd/queues/controller.go index 9ee9d464..d20f6d7c 100755 --- a/src/frontend/xcecli/cmd/queues/controller.go +++ b/src/frontend/xcecli/cmd/queues/controller.go @@ -1,239 +1,237 @@ -package queues - -import ( - "encoding/json" - "fmt" - "strings" - "xcecli/common/request" - "xcecli/config" - "xcecli/models" - - "github.com/spf13/cobra" -) - -// QueueController queue controller -type QueueController struct { - configManager *config.ConfigManager - action string - queueNames []string - comment string - allQueues bool -} - -// NewQueueController create queue controller -func NewQueueController(configManager *config.ConfigManager, action string) *QueueController { - return &QueueController{ - configManager: configManager, - action: action, - queueNames: []string{}, - } -} - -// Run execute queue control -func (c *QueueController) RunQueuesControl() error { - // get server info and verify authentication - serverInfo, _ := c.configManager.GetServerInfo() - // create API client - apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) - - // handle all queues case - if c.allQueues { - c.queueNames = []string{"all"} - } - - // if no queue names are specified, return an error - if len(c.queueNames) == 0 { - return fmt.Errorf("must specify queue names or use --all option") - } - - // create request - req := models.QueueControlRequest{ - ControlAction: c.action, - QueueName: c.queueNames, - Comment: c.comment, - } - - // convert request to JSON bytes - reqData, err := json.Marshal(req) - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to encode request data: [%s]", errMsg) - } - - // send request - var resp models.QueueControlResponse - err = apiClient.PostRequestToAPIServer( - serverInfo.Token, - serverInfo.CACert, - "/xce/v1/queues/control", - reqData, - &resp, - ) - - if err != nil { - return err - } - - if len(c.queueNames) == 1 && c.queueNames[0] == "all" { - for _, data := range resp.Data.Accepted { - fmt.Printf("queue <%s> is being <%s>\n", data.QueueName, c.action) - } - for _, queue := range resp.Data.Failed { - fmt.Printf("queue <%s> cannot be <%s> for reason <%s>\n", queue.QueueName, c.action, queue.Msg) - } - } else { - for _, data := range resp.Data.Accepted { - fmt.Printf("queue <%s> is being <%s>\n", data.QueueName, c.action) - } - for _, queue := range resp.Data.Failed { - fmt.Printf("queue <%s> cannot be <%s> for reason <%s>\n", queue.QueueName, c.action, queue.Msg) - } - } - return nil -} - -// Parse parse command line arguments -func (c *QueueController) ParseCmdArgs(cmd *cobra.Command, args []string) error { - // handle queue name parameters - if len(args) > 0 { - // Split comma-separated queue names - for _, arg := range args { - queues := strings.Split(arg, ",") - c.queueNames = append(c.queueNames, queues...) - } - } - - return nil -} - -// NewOpenQueuesCmd create open queues command -func NewOpenQueuesCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewQueueController(configManager, "open") - - cmd := &cobra.Command{ - Use: "open", - Short: "Open queues", - Long: `Open queues by queue name. -Opens queues to accept new job submissions. - -Examples: - xce queues open normal priority # open specific queues - xce queues open --all # open all queues`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunQueuesControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues") - flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") - - return cmd -} - -// NewCloseQueuesCmd create close queues command -func NewCloseQueuesCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewQueueController(configManager, "close") - - cmd := &cobra.Command{ - Use: "close", - Short: "Close queues", - Long: `Close queues by queue name. -Closes queues to prevent new job submissions. - -Examples: - xce queues close normal priority # close specific queues - xce queues close --all # close all queues`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunQueuesControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues") - flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") - - return cmd -} - -// NewActiveQueuesCmd create active queues command -func NewActiveQueuesCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewQueueController(configManager, "active") - - cmd := &cobra.Command{ - Use: "active", - Short: "Active queues", - Long: `Active queues by queue name. - -Examples: - xce queues active # queues_name status close/d`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunQueuesControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues") - flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") - - return cmd -} - -func NewDeactiveQueuesCmd(configManager *config.ConfigManager) *cobra.Command { - controller := NewQueueController(configManager, "deactive") - - cmd := &cobra.Command{ - Use: "deactive", - Short: "Deactive queues", - Long: `Deactive queues by queue name. - -Examples: - xce queues deactive # queues_name status close/d`, - RunE: func(cmd *cobra.Command, args []string) error { - if err := controller.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := controller.RunQueuesControl(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // add control options - flags := cmd.Flags() - flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues") - flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") - - return cmd -} +package queues + +import ( + "encoding/json" + "fmt" + "strings" + "xcecli/common/request" + "xcecli/config" + "xcecli/models" + + "github.com/spf13/cobra" +) + +// QueueController queue controller +type QueueController struct { + configManager *config.ConfigManager + action string + queueNames []string + comment string + allQueues bool +} + +// NewQueueController create queue controller +func NewQueueController(configManager *config.ConfigManager, action string) *QueueController { + return &QueueController{ + configManager: configManager, + action: action, + queueNames: []string{}, + } +} + +// Run execute queue control +func (c *QueueController) RunQueuesControl() error { + // get server info and verify authentication + serverInfo, _ := c.configManager.GetServerInfo() + // create API client + apiClient := request.NewAPIClient(serverInfo.URL) + + // handle all queues case + if c.allQueues { + c.queueNames = []string{"all"} + } + + // if no queue names are specified, return an error + if len(c.queueNames) == 0 { + return fmt.Errorf("must specify queue names or use --all option") + } + + // create request + req := models.QueueControlRequest{ + ControlAction: c.action, + QueueName: c.queueNames, + Comment: c.comment, + } + + // convert request to JSON bytes + reqData, err := json.Marshal(req) + if err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to encode request data: [%s]", errMsg) + } + + // send request + var resp models.QueueControlResponse + err = apiClient.PostRequestToAPIServer( + serverInfo.Token, + "/xce/v1/queues/control", + reqData, + &resp, + ) + + if err != nil { + return err + } + + if len(c.queueNames) == 1 && c.queueNames[0] == "all" { + for _, data := range resp.Data.Accepted { + fmt.Printf("queue <%s> is being <%s>\n", data.QueueName, c.action) + } + for _, queue := range resp.Data.Failed { + fmt.Printf("queue <%s> cannot be <%s> for reason <%s>\n", queue.QueueName, c.action, queue.Msg) + } + } else { + for _, data := range resp.Data.Accepted { + fmt.Printf("queue <%s> is being <%s>\n", data.QueueName, c.action) + } + for _, queue := range resp.Data.Failed { + fmt.Printf("queue <%s> cannot be <%s> for reason <%s>\n", queue.QueueName, c.action, queue.Msg) + } + } + return nil +} + +// Parse parse command line arguments +func (c *QueueController) ParseCmdArgs(cmd *cobra.Command, args []string) error { + // handle queue name parameters + if len(args) > 0 { + // Split comma-separated queue names + for _, arg := range args { + queues := strings.Split(arg, ",") + c.queueNames = append(c.queueNames, queues...) + } + } + + return nil +} + +// NewOpenQueuesCmd create open queues command +func NewOpenQueuesCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewQueueController(configManager, "open") + + cmd := &cobra.Command{ + Use: "open", + Short: "Open queues", + Long: `Open queues by queue name. +Opens queues to accept new job submissions. + +Examples: + xce queues open normal priority # open specific queues + xce queues open --all # open all queues`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunQueuesControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues") + flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") + + return cmd +} + +// NewCloseQueuesCmd create close queues command +func NewCloseQueuesCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewQueueController(configManager, "close") + + cmd := &cobra.Command{ + Use: "close", + Short: "Close queues", + Long: `Close queues by queue name. +Closes queues to prevent new job submissions. + +Examples: + xce queues close normal priority # close specific queues + xce queues close --all # close all queues`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunQueuesControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues") + flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") + + return cmd +} + +// NewActiveQueuesCmd create active queues command +func NewActiveQueuesCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewQueueController(configManager, "active") + + cmd := &cobra.Command{ + Use: "active", + Short: "Active queues", + Long: `Active queues by queue name. + +Examples: + xce queues active # queues_name status close/d`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunQueuesControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues") + flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") + + return cmd +} + +func NewDeactiveQueuesCmd(configManager *config.ConfigManager) *cobra.Command { + controller := NewQueueController(configManager, "deactive") + + cmd := &cobra.Command{ + Use: "deactive", + Short: "Deactive queues", + Long: `Deactive queues by queue name. + +Examples: + xce queues deactive # queues_name status close/d`, + RunE: func(cmd *cobra.Command, args []string) error { + if err := controller.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := controller.RunQueuesControl(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // add control options + flags := cmd.Flags() + flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues") + flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") + + return cmd +} diff --git a/src/frontend/xcecli/cmd/queues/list.go b/src/frontend/xcecli/cmd/queues/list.go index ac3786cb..7cc6cee9 100755 --- a/src/frontend/xcecli/cmd/queues/list.go +++ b/src/frontend/xcecli/cmd/queues/list.go @@ -1,199 +1,198 @@ -package queues - -import ( - "fmt" - "strings" - "xcecli/common/output" - "xcecli/common/request" - "xcecli/config" - logger "xcecli/config" - queue_filter "xcecli/filters" - - "xcecli/models" - - "github.com/spf13/cobra" - "go.uber.org/zap" -) - -var ( - user string - queue string - fields string -) - -type QueuesQuerier struct { - configManager *config.ConfigManager - filter *queue_filter.QueuesFilter - fields string - jsonOutput bool - noHeader bool - verbose bool - wideFormat bool // wide format -} - -func NewQueuesQuerier(configManager *config.ConfigManager) *QueuesQuerier { - return &QueuesQuerier{ - configManager: configManager, - filter: &queue_filter.QueuesFilter{}, - } -} - -// Parse parses command line arguments, simplified version that doesn't need to process flags already bound to filter -func (qq *QueuesQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error { - // No additional processing needed, all parameters are bound in flags - if len(args) > 0 { - qq.filter.QueueName = strings.Join(args, ",") - } - - // Define mutual exclusions for verbose flag - if qq.verbose { - conflicts := make([]string, 0) - - if qq.wideFormat { - conflicts = append(conflicts, "--wided (-w)") - } - - if qq.fields != "" { - conflicts = append(conflicts, "--output (-o)") - } - - if len(conflicts) > 0 { - return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", - strings.Join(conflicts, ", "), cmd.UsageString()) - } - } - return nil -} - -func NewListQueuesCmd(configManager *config.ConfigManager) *cobra.Command { - queuesQuerier := NewQueuesQuerier(configManager) - - cmd := &cobra.Command{ - Use: "list", - Short: "Query queue info", - Long: `Query queue info, support filter by queue. -example: - xce queues list # query all queues - xce queues list -l # query all queues with detail - xce queues list -w # list queues with wide format`, - - RunE: func(cmd *cobra.Command, args []string) error { - if err := queuesQuerier.ParseCmdArgs(cmd, args); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - if err := queuesQuerier.RunQueuesList(); err != nil { - fmt.Println(strings.ToLower(err.Error())) - return nil - } - return nil - }, - } - - // Add command line parameters - flags := cmd.Flags() - flags.StringVarP(&user, "user", "u", "", "filter by user") - flags.StringVarP(&queuesQuerier.filter.QueueName, "queue", "q", "", "filter by queue name") - flags.StringVarP(&queuesQuerier.fields, "output", "o", "", "output fields for queues") - flags.BoolVarP(&queuesQuerier.wideFormat, "wided", "w", false, "output in wide format") - flags.BoolVarP(&queuesQuerier.verbose, "verbose", "l", false, "show queue detail") - flags.BoolVarP(&queuesQuerier.jsonOutput, "json", "", false, "output in JSON format") - flags.BoolVarP(&queuesQuerier.noHeader, "noheader", "n", false, "output no header") - return cmd -} - -func (qq *QueuesQuerier) RunQueuesList() error { - // Get configuration - - // Get server information - serverInfo, _ := qq.configManager.GetServerInfo() - // create API client - apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) - - // Create query parameters - params := make(map[string]string) - - // Get filter string - filterStr, err := qq.filter.HandleQueueFilter() - - if err != nil { - errMsg := request.ExtractErrorMessage(err) - return fmt.Errorf("failed to get filter: [%s]", errMsg) - } else if filterStr != "" { - params["filter"] = filterStr - } - - // Field mapping - fieldsMap := models.QueueFieldsMap - - if qq.verbose && qq.fields == "" { - for header := range fieldsMap { - qq.fields += header + "," - } - qq.fields = strings.TrimSuffix(qq.fields, ",") - } - // Process field selection - if qq.fields != "" { - qq.fields = request.CleanFields(qq.fields) - mappedFields, err := request.CheckAndMapFields(qq.fields, fieldsMap) - if err != nil { - return err - } - params["fields"] = strings.Join(mappedFields, ",") - } - - // create page printer - pagePrinter := output.NewPagePrinter(output.DataTypeQueues, output.WithFields(qq.fields), - output.WithFormatJson(qq.jsonOutput), output.WithNoHeader(qq.noHeader), - output.WithVerbose(qq.verbose), output.WithWideFormat(qq.wideFormat)) - - // Record if the first page has been processed - isFirstPage := true - - var ( - resp models.QueuesResponse - loopCount int - pageToken string - ) - - // Loop to get all pages of data - for { - if loopCount > 0 && pageToken == "" { - break - } - - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/queues", &resp, params) - if err != nil { - return err - } - - if len(resp.Data.Rows) == 0 { - if isFirstPage { - fmt.Println("no queue found") - return nil - } - // - } - // Process the first page - if isFirstPage { - // Start printing process - pagePrinter.StartPrinting() - // Prepare for printing, calculate column width and table header, etc. - pagePrinter.PrepareForPrinting(&resp) - isFirstPage = false - } - // Print the current page - pagePrinter.PrintPage(&resp) - - // Update page token and loop counter - pageToken = resp.Data.Page_token - loopCount++ - } - - // Finish printing - pagePrinter.Finish() - - logger.Debug("queues response completed", zap.Int("pages", loopCount)) - return nil -} +package queues + +import ( + "fmt" + "strings" + "xcecli/common/output" + "xcecli/common/request" + "xcecli/config" + logger "xcecli/config" + queue_filter "xcecli/filters" + + "xcecli/models" + + "github.com/spf13/cobra" + "go.uber.org/zap" +) + +var ( + user string + queue string + fields string +) + +type QueuesQuerier struct { + configManager *config.ConfigManager + filter *queue_filter.QueuesFilter + fields string + jsonOutput bool + noHeader bool + verbose bool + wideFormat bool // wide format +} + +func NewQueuesQuerier(configManager *config.ConfigManager) *QueuesQuerier { + return &QueuesQuerier{ + configManager: configManager, + filter: &queue_filter.QueuesFilter{}, + } +} + +// Parse parses command line arguments, simplified version that doesn't need to process flags already bound to filter +func (qq *QueuesQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error { + // No additional processing needed, all parameters are bound in flags + if len(args) > 0 { + qq.filter.QueueName = strings.Join(args, ",") + } + + // Define mutual exclusions for verbose flag + if qq.verbose { + conflicts := make([]string, 0) + + if qq.wideFormat { + conflicts = append(conflicts, "--wided (-w)") + } + + if qq.fields != "" { + conflicts = append(conflicts, "--output (-o)") + } + + if len(conflicts) > 0 { + return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", + strings.Join(conflicts, ", "), cmd.UsageString()) + } + } + return nil +} + +func NewListQueuesCmd(configManager *config.ConfigManager) *cobra.Command { + queuesQuerier := NewQueuesQuerier(configManager) + + cmd := &cobra.Command{ + Use: "list", + Short: "Query queue info", + Long: `Query queue info, support filter by queue. +example: + xce queues list # query all queues + xce queues list -l # query all queues with detail + xce queues list -w # list queues with wide format`, + + RunE: func(cmd *cobra.Command, args []string) error { + if err := queuesQuerier.ParseCmdArgs(cmd, args); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + if err := queuesQuerier.RunQueuesList(); err != nil { + fmt.Println(strings.ToLower(err.Error())) + return nil + } + return nil + }, + } + + // Add command line parameters + flags := cmd.Flags() + flags.StringVarP(&user, "user", "u", "", "filter by user") + flags.StringVarP(&queuesQuerier.filter.QueueName, "queue", "q", "", "filter by queue name") + flags.StringVarP(&queuesQuerier.fields, "output", "o", "", "output fields for queues") + flags.BoolVarP(&queuesQuerier.wideFormat, "wided", "w", false, "output in wide format") + flags.BoolVarP(&queuesQuerier.verbose, "verbose", "l", false, "show queue detail") + flags.BoolVarP(&queuesQuerier.jsonOutput, "json", "", false, "output in JSON format") + flags.BoolVarP(&queuesQuerier.noHeader, "noheader", "n", false, "output no header") + return cmd +} + +func (qq *QueuesQuerier) RunQueuesList() error { + // Get configuration + + // Get server information + serverInfo, _ := qq.configManager.GetServerInfo() + // create API client + apiClient := request.NewAPIClient(serverInfo.URL) + + // Create query parameters + params := make(map[string]string) + + // Get filter string + filterStr, err := qq.filter.HandleQueueFilter() + + if err != nil { + errMsg := request.ExtractErrorMessage(err) + return fmt.Errorf("failed to get filter: [%s]", errMsg) + } else if filterStr != "" { + params["filter"] = filterStr + } + + // Field mapping + fieldsMap := models.QueueFieldsMap + + if qq.verbose && qq.fields == "" { + for header := range fieldsMap { + qq.fields += header + "," + } + qq.fields = strings.TrimSuffix(qq.fields, ",") + } + // Process field selection + if qq.fields != "" { + qq.fields = request.CleanFields(qq.fields) + mappedFields, err := request.CheckAndMapFields(qq.fields, fieldsMap) + if err != nil { + return err + } + params["fields"] = strings.Join(mappedFields, ",") + } + + // create page printer + pagePrinter := output.NewPagePrinter(output.DataTypeQueues, output.WithFields(qq.fields), + output.WithFormatJson(qq.jsonOutput), output.WithNoHeader(qq.noHeader), + output.WithVerbose(qq.verbose), output.WithWideFormat(qq.wideFormat)) + + // Record if the first page has been processed + isFirstPage := true + + var ( + resp models.QueuesResponse + loopCount int + pageToken string + ) + + // Loop to get all pages of data + for { + if loopCount > 0 && pageToken == "" { + break + } + + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/queues", &resp, params) + if err != nil { + return err + } + + if len(resp.Data.Rows) == 0 { + if isFirstPage { + fmt.Println("no queue found") + return nil + } + // + } + // Process the first page + if isFirstPage { + // Start printing process + pagePrinter.StartPrinting() + // Prepare for printing, calculate column width and table header, etc. + pagePrinter.PrepareForPrinting(&resp) + isFirstPage = false + } + // Print the current page + pagePrinter.PrintPage(&resp) + + // Update page token and loop counter + pageToken = resp.Data.Page_token + loopCount++ + } + + // Finish printing + pagePrinter.Finish() + + logger.Debug("queues response completed", zap.Int("pages", loopCount)) + return nil +} diff --git a/src/frontend/xcecli/cmd/resources/list.go b/src/frontend/xcecli/cmd/resources/list.go index 6ebd386a..9bbcea1c 100644 --- a/src/frontend/xcecli/cmd/resources/list.go +++ b/src/frontend/xcecli/cmd/resources/list.go @@ -135,7 +135,6 @@ func (rq *ResourcesDefQuerier) RunResourcesDefList() error { serverInfo, _ := rq.configManager.GetServerInfo() // create API client apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // Create query parameters params := make(map[string]string) @@ -188,7 +187,7 @@ func (rq *ResourcesDefQuerier) RunResourcesDefList() error { if loopCount > 0 && pageToken == "" { break } - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/resources/definition", &resp, params) + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/resources/definition", &resp, params) if err != nil { return err } @@ -263,7 +262,6 @@ func (rq *ResourcesDistQuerier) RunResourcesDistList() error { serverInfo, _ := rq.configManager.GetServerInfo() // create API client apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // Create query parameters params := make(map[string]string) @@ -323,7 +321,7 @@ func (rq *ResourcesDistQuerier) RunResourcesDistList() error { if loopCount > 0 && pageToken == "" { break } - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/resources/distribution", &resp, params) + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/resources/distribution", &resp, params) if err != nil { return err } diff --git a/src/frontend/xcecli/cmd/roles/addrole.go b/src/frontend/xcecli/cmd/roles/addrole.go index 3c5ffdba..3b107550 100644 --- a/src/frontend/xcecli/cmd/roles/addrole.go +++ b/src/frontend/xcecli/cmd/roles/addrole.go @@ -206,7 +206,6 @@ func (rs *RoleSubmitter) RunRolesAdd() error { serverInfo, _ := rs.configManager.GetServerInfo() // create the API client and submit the role apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // build the role request roleReq, err := rs.BuildRoleRequest() @@ -223,7 +222,6 @@ func (rs *RoleSubmitter) RunRolesAdd() error { var resp models.CreateRoleResponse err = apiClient.PostRequestToAPIServer( serverInfo.Token, - serverInfo.CACert, "/xce/v1/roles/create", reqData, &resp, @@ -248,7 +246,6 @@ func (rs *RoleSubmitter) RunPermsAdd() error { serverInfo, _ := rs.configManager.GetServerInfo() // create the API client and submit the role apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // build the role request roleReq, err := rs.BuildRoleRequest() @@ -267,7 +264,6 @@ func (rs *RoleSubmitter) RunPermsAdd() error { var resp models.CreateRoleResponse err = apiClient.PostRequestToAPIServer( serverInfo.Token, - serverInfo.CACert, url, reqData, &resp, diff --git a/src/frontend/xcecli/cmd/roles/deleterole.go b/src/frontend/xcecli/cmd/roles/deleterole.go index 7ba4603e..cc8ddbb9 100644 --- a/src/frontend/xcecli/cmd/roles/deleterole.go +++ b/src/frontend/xcecli/cmd/roles/deleterole.go @@ -65,13 +65,11 @@ func (dr *DeleteRole) RunDeleteRoleInfo() error { serverInfo, _ := dr.configManager.GetServerInfo() // create the API client and submit the user apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) url := fmt.Sprintf("/xce/v1/roles/%s", dr.roleName) var resp request.APIResponse err := apiClient.DeleteRequestFromAPIServer( serverInfo.Token, - serverInfo.CACert, url, []byte(nil), &resp, diff --git a/src/frontend/xcecli/cmd/roles/list.go b/src/frontend/xcecli/cmd/roles/list.go index 7c037769..52ffeae0 100644 --- a/src/frontend/xcecli/cmd/roles/list.go +++ b/src/frontend/xcecli/cmd/roles/list.go @@ -88,7 +88,6 @@ func (rq *RolesQuerier) RunRolesInfo(commandType string) error { // create API client apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // Create query parameters params := make(map[string]string) @@ -156,7 +155,7 @@ func (rq *RolesQuerier) RunRolesInfo(commandType string) error { break } resp = models.RolesResponse{} // reset to avoid stale rows from previous page - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/roles", &resp, params) + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/roles", &resp, params) if err != nil { return err } diff --git a/src/frontend/xcecli/cmd/users/assignrole.go b/src/frontend/xcecli/cmd/users/assignrole.go index d0e664b4..0daeeea6 100644 --- a/src/frontend/xcecli/cmd/users/assignrole.go +++ b/src/frontend/xcecli/cmd/users/assignrole.go @@ -80,7 +80,6 @@ func (us *UserSubmitter) RunUserAssignRole() error { serverInfo, _ := us.configManager.GetServerInfo() // create the API client and submit the user apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // build the user request userReq, err := us.BuildUserRequest() @@ -98,7 +97,6 @@ func (us *UserSubmitter) RunUserAssignRole() error { var resp models.UserAssignRoleResponse err = apiClient.PostRequestToAPIServer( serverInfo.Token, - serverInfo.CACert, url, reqData, &resp, diff --git a/src/frontend/xcecli/cmd/users/list.go b/src/frontend/xcecli/cmd/users/list.go index 3d743d73..fc2b6ad8 100644 --- a/src/frontend/xcecli/cmd/users/list.go +++ b/src/frontend/xcecli/cmd/users/list.go @@ -104,13 +104,12 @@ func (urq *UserRolesQuerier) RunCurrentUserInfo() error { // create API client apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // Create a response structure for current user info var currentUserResp models.UserResponse // You'll need to define this model // Make API request to get current user info using the token - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, "/xce/v1/users/current", ¤tUserResp, nil) + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/xce/v1/users/current", ¤tUserResp, nil) if err != nil { return fmt.Errorf("failed to get current user info: %w", err) } @@ -129,7 +128,6 @@ func (urq *UserRolesQuerier) RunUserRolesInfo(commandType string) error { // create API client apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // Create query parameters params := make(map[string]string) @@ -191,7 +189,7 @@ func (urq *UserRolesQuerier) RunUserRolesInfo(commandType string) error { break } resp = models.UserRolesResponse{} // reset to avoid stale rows from previous page - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/users", &resp, params) + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/users", &resp, params) if err != nil { return err } diff --git a/src/frontend/xcecli/cmd/users/perm4user.go b/src/frontend/xcecli/cmd/users/perm4user.go index 95b8494e..dc93d91c 100644 --- a/src/frontend/xcecli/cmd/users/perm4user.go +++ b/src/frontend/xcecli/cmd/users/perm4user.go @@ -84,7 +84,6 @@ func (up *Perms4User) RunUserPermsInfo() error { // create API client apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) // Create query parameters params := make(map[string]string) @@ -116,7 +115,7 @@ func (up *Perms4User) RunUserPermsInfo() error { // Loop to get all pages of data url := "/xce/v1/users/" + up.userName + "/permissions" - err = apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, url, &resp, nil) + err = apiClient.GetRequestFromAPIServer(serverInfo.Token, "", url, &resp, nil) if err != nil { return err } diff --git a/src/frontend/xcecli/cmd/users/unassignrole.go b/src/frontend/xcecli/cmd/users/unassignrole.go index 09931dbd..768ac585 100644 --- a/src/frontend/xcecli/cmd/users/unassignrole.go +++ b/src/frontend/xcecli/cmd/users/unassignrole.go @@ -73,13 +73,11 @@ func (uu *UnassignUser) RunDeleteUserRoleInfo() error { serverInfo, _ := uu.configManager.GetServerInfo() // create the API client and submit the user apiClient := request.NewAPIClient(serverInfo.URL) - apiClient.SetRootCAs(serverInfo.CACert) url := fmt.Sprintf("/xce/v1/users/%s/%s", uu.userName, uu.roleName) var resp request.APIResponse err := apiClient.DeleteRequestFromAPIServer( serverInfo.Token, - serverInfo.CACert, url, []byte(nil), &resp, diff --git a/src/frontend/xcecli/common/request/request.go b/src/frontend/xcecli/common/request/request.go index c3689c09..615af317 100755 --- a/src/frontend/xcecli/common/request/request.go +++ b/src/frontend/xcecli/common/request/request.go @@ -1,8 +1,6 @@ package request import ( - "crypto/tls" - "crypto/x509" "encoding/json" "fmt" @@ -45,34 +43,6 @@ func NewAPIClient(baseURL string) *APIClient { return c } -func (c *APIClient) SetRootCAs(certPath string) { - if !strings.HasPrefix(c.BaseURL, "https://") { - return - } - if certPath == "" { - c.Client.SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true}) //nolint:gosec - return - } - pool := loadRootCAs(certPath) - c.Client.SetTLSClientConfig(&tls.Config{RootCAs: pool}) -} - -// loadRootCAs load the root certificate -func loadRootCAs(certPath string) *x509.CertPool { - certPool := x509.NewCertPool() - - cert, err := os.ReadFile(certPath) - if err != nil { - logger.Error("failed to read the cert file") - return nil - } - if ok := certPool.AppendCertsFromPEM(cert); !ok { - logger.Error("failed to append the cert") - return nil - } - return certPool -} - // Logon perform the login operation func (c *APIClient) Logon(username, password string) (*models.LogonResponse, error) { var loginResp models.LogonResponse @@ -100,13 +70,8 @@ func (c *APIClient) Logon(username, password string) (*models.LogonResponse, err return &loginResp, nil } -func (c *APIClient) GetRequestFromAPIServer(token, pageToken, certPath, dst string, resp any, params map[string]string) error { +func (c *APIClient) GetRequestFromAPIServer(token, pageToken, dst string, resp any, params map[string]string) error { baseURL := c.BaseURL - if strings.HasPrefix(baseURL, "https://") { - c.Client.SetTLSClientConfig(&tls.Config{ - RootCAs: loadRootCAs(certPath), - }) - } if idx := strings.Index(baseURL, dst); idx != -1 { baseURL = baseURL[:idx] } @@ -271,13 +236,8 @@ func (c *APIClient) PostFile(token string, files []string) (*models.UploadRespon } // Post sends a POST request to the given URL with the given data -func (c *APIClient) PostRequestToAPIServer(token, certPath, dst string, data []byte, resp interface{}) error { +func (c *APIClient) PostRequestToAPIServer(token, dst string, data []byte, resp interface{}) error { baseURL := c.BaseURL - if strings.HasPrefix(baseURL, "https://") && certPath != "" { - c.Client.SetTLSClientConfig(&tls.Config{ - RootCAs: loadRootCAs(certPath), - }) - } if idx := strings.Index(baseURL, dst); idx != -1 { baseURL = baseURL[:idx] } @@ -306,13 +266,8 @@ func (c *APIClient) PostRequestToAPIServer(token, certPath, dst string, data []b return nil } -func (c *APIClient) DeleteRequestFromAPIServer(token, certPath, dst string, data []byte, resp interface{}) error { +func (c *APIClient) DeleteRequestFromAPIServer(token, dst string, data []byte, resp interface{}) error { baseURL := c.BaseURL - if strings.HasPrefix(baseURL, "https://") && certPath != "" { - c.Client.SetTLSClientConfig(&tls.Config{ - RootCAs: loadRootCAs(certPath), - }) - } if idx := strings.Index(baseURL, dst); idx != -1 { baseURL = baseURL[:idx] } diff --git a/src/frontend/xcecli/common/request/request_test.go b/src/frontend/xcecli/common/request/request_test.go index ab597114..b8c53ca0 100644 --- a/src/frontend/xcecli/common/request/request_test.go +++ b/src/frontend/xcecli/common/request/request_test.go @@ -2,36 +2,6 @@ package request import "testing" -// TestSetRootCAs_HTTPNoOp verifies HTTP clients are not affected. -func TestSetRootCAs_HTTPNoOp(t *testing.T) { - c := NewAPIClient("http://localhost:8080") - c.SetRootCAs("/some/cert.pem") // must not panic -} - -// TestSetRootCAs_HTTPSEmptyCert verifies InsecureSkipVerify is set when no cert provided. -func TestSetRootCAs_HTTPSEmptyCert(t *testing.T) { - c := NewAPIClient("https://localhost:8443") - c.SetRootCAs("") // no cert → insecure - // If we reach here without panic, the call succeeded. - // Behavioral correctness is verified by TestBuildTLSLogic below. -} - -// TestBuildTLSLogic tests the cert-path branching logic directly. -func TestBuildTLSLogic(t *testing.T) { - // empty cert → insecure - if certPath := ""; certPath != "" { - t.Error("logic error in test") - } - // non-empty cert → not insecure (even if file missing, InsecureSkipVerify=false) - insecure := func(certPath string) bool { return certPath == "" } - if !insecure("") { - t.Error("empty certPath should be insecure") - } - if insecure("/some/cert.pem") { - t.Error("non-empty certPath should not be insecure") - } -} - func TestAuthHeaderConsistency(t *testing.T) { token := "test-token-abc" headers := map[string]string{ diff --git a/src/frontend/xcecli/config/config.go b/src/frontend/xcecli/config/config.go index 7c729750..b7e45c46 100755 --- a/src/frontend/xcecli/config/config.go +++ b/src/frontend/xcecli/config/config.go @@ -1,207 +1,206 @@ -package config - -import ( - "encoding/json" - "fmt" - "os" - "os/user" - "path" - "sort" - "time" -) - -type APIServerInfo struct { - // Name string `json:"name"` - URL string `json:"url"` - Token string `json:"token"` - Path string `json:"path"` - CreateTime int64 `json:"createtime"` - CACert string `json:"cacert,omitempty"` - TokenExpireTime int64 `json:"token_expire_time"` - //Version string `json:"version,omitempty"` -} - -// APIServerInfoArray is used to sort the APIServerInfo by CreateTime -type APIServerInfoArray []APIServerInfo - -func (a APIServerInfoArray) Len() int { return len(a) } -func (a APIServerInfoArray) Swap(i, j int) { a[i], a[j] = a[j], a[i] } -func (a APIServerInfoArray) Less(i, j int) bool { return a[i].CreateTime > a[j].CreateTime } - -type Config struct { - APIServerInfo APIServerInfoArray `json:"servers"` -} - -// ConfigManager is used to manage the config -type ConfigManager struct { - config *Config - configPath string -} - -// NewConfigManager is used to create the config manager -func NewConfigManager() (*ConfigManager, error) { - usr, err := user.Current() - if err != nil { - return nil, fmt.Errorf("failed to get the user info: [%v]", err) - } - - configDir := path.Join(usr.HomeDir, ".xce") - configPath := path.Join(configDir, "config.json") - - // create the config directory if it does not exist - if err := os.MkdirAll(configDir, 0700); err != nil { - return nil, fmt.Errorf("failed to create the config directory: [%v]", err) - } - - cm := &ConfigManager{ - configPath: configPath, - } - - return cm, nil -} - -// GetConfig is used to get the config -func (cm *ConfigManager) GetConfig() (*Config, error) { - if cm == nil { - return nil, fmt.Errorf("the config manager is not initialized") - } - - if cm.config != nil { - return cm.config, nil - } - - config, err := cm.loadConfig() - if err != nil { - return nil, err - } - - cm.config = config - return config, nil -} - -// SaveConfig is used to save the config -func (cm *ConfigManager) SaveConfig(config *Config) error { - // sort the APIServerInfoArray by CreateTime - sort.Sort(config.APIServerInfo) - data, err := json.MarshalIndent(config, "", " ") - if err != nil { - return err - } - - dir := path.Dir(cm.configPath) - if err := os.MkdirAll(dir, 0700); err != nil { - return err - } - - if err := os.WriteFile(cm.configPath, data, 0600); err != nil { - return err - } - - cm.config = config - return nil -} - -// loadConfig is used to load the config from the file -func (cm *ConfigManager) loadConfig() (*Config, error) { - // if the config file does not exist, create the default config - if _, err := os.Stat(cm.configPath); os.IsNotExist(err) { - defaultConfig := &Config{ - APIServerInfo: make([]APIServerInfo, 0), - } - // save the default config - if err := cm.SaveConfig(defaultConfig); err != nil { - return nil, fmt.Errorf("failed to create the default config: [%v]", err) - } - return defaultConfig, nil - } - - // read the config file - data, err := os.ReadFile(cm.configPath) - if err != nil { - return nil, fmt.Errorf("failed to read the config file: [%v]", err) - } - - // parse the config file - var config Config - if err := json.Unmarshal(data, &config); err != nil { - return nil, fmt.Errorf("failed to parse the config file: [%s,%v]", cm.configPath, err) - } - - return &config, nil -} - -// Save is used to save the config -func (c *Config) Save() error { - usr, err := user.Current() - if err != nil { - return err - } - - // get the config directory - configDir := path.Join(usr.HomeDir, ".xce") - configPath := path.Join(configDir, "config.json") - - // create the config directory if it does not exist - if err := os.MkdirAll(configDir, 0700); err != nil { - return err - } - - // marshal the config - data, err := json.MarshalIndent(c, "", " ") - if err != nil { - return err - } - - // write the config to the file - return os.WriteFile(configPath, data, 0600) -} - -// IsAuthenticated check if the user has been authenticated by the API server -// if not, return an error -func (cm *ConfigManager) IsAuthenticated() (*APIServerInfo, error) { - cfg, err := cm.GetConfig() - if err != nil { - return nil, fmt.Errorf("failed to get config: [%v]", err) - } - - if len(cfg.APIServerInfo) == 0 { - return nil, fmt.Errorf("you are not logged in, please log in first") - } - - serverInfo := &cfg.APIServerInfo[0] - if serverInfo.URL == "" { - return nil, fmt.Errorf("no default API Server is set, please set a default server") - } - - if serverInfo.Token == "" { - return nil, fmt.Errorf("you are not logged in, please log in first") - } - - // Use the token expire time from server configuration - expireTime := serverInfo.TokenExpireTime - if expireTime == 0 { - expireTime = 86400 // Default to 24 hours if not set - } - if time.Now().Unix()-serverInfo.CreateTime > expireTime { - return nil, fmt.Errorf("your login has expired, please log in again") - } - - return serverInfo, nil -} - -// GetServerInfo -func (cm *ConfigManager) GetServerInfo() (*APIServerInfo, error) { - cfg, err := cm.GetConfig() - if err != nil { - return nil, fmt.Errorf("failed to get config: [%v]", err) - } - - if len(cfg.APIServerInfo) == 0 || cfg.APIServerInfo[0].URL == "" { - return nil, fmt.Errorf("not found the API Server information") - } - - return &cfg.APIServerInfo[0], nil -} +package config + +import ( + "encoding/json" + "fmt" + "os" + "os/user" + "path" + "sort" + "time" +) + +type APIServerInfo struct { + // Name string `json:"name"` + URL string `json:"url"` + Token string `json:"token"` + Path string `json:"path"` + CreateTime int64 `json:"createtime"` + TokenExpireTime int64 `json:"token_expire_time"` + //Version string `json:"version,omitempty"` +} + +// APIServerInfoArray is used to sort the APIServerInfo by CreateTime +type APIServerInfoArray []APIServerInfo + +func (a APIServerInfoArray) Len() int { return len(a) } +func (a APIServerInfoArray) Swap(i, j int) { a[i], a[j] = a[j], a[i] } +func (a APIServerInfoArray) Less(i, j int) bool { return a[i].CreateTime > a[j].CreateTime } + +type Config struct { + APIServerInfo APIServerInfoArray `json:"servers"` +} + +// ConfigManager is used to manage the config +type ConfigManager struct { + config *Config + configPath string +} + +// NewConfigManager is used to create the config manager +func NewConfigManager() (*ConfigManager, error) { + usr, err := user.Current() + if err != nil { + return nil, fmt.Errorf("failed to get the user info: [%v]", err) + } + + configDir := path.Join(usr.HomeDir, ".xce") + configPath := path.Join(configDir, "config.json") + + // create the config directory if it does not exist + if err := os.MkdirAll(configDir, 0700); err != nil { + return nil, fmt.Errorf("failed to create the config directory: [%v]", err) + } + + cm := &ConfigManager{ + configPath: configPath, + } + + return cm, nil +} + +// GetConfig is used to get the config +func (cm *ConfigManager) GetConfig() (*Config, error) { + if cm == nil { + return nil, fmt.Errorf("the config manager is not initialized") + } + + if cm.config != nil { + return cm.config, nil + } + + config, err := cm.loadConfig() + if err != nil { + return nil, err + } + + cm.config = config + return config, nil +} + +// SaveConfig is used to save the config +func (cm *ConfigManager) SaveConfig(config *Config) error { + // sort the APIServerInfoArray by CreateTime + sort.Sort(config.APIServerInfo) + data, err := json.MarshalIndent(config, "", " ") + if err != nil { + return err + } + + dir := path.Dir(cm.configPath) + if err := os.MkdirAll(dir, 0700); err != nil { + return err + } + + if err := os.WriteFile(cm.configPath, data, 0600); err != nil { + return err + } + + cm.config = config + return nil +} + +// loadConfig is used to load the config from the file +func (cm *ConfigManager) loadConfig() (*Config, error) { + // if the config file does not exist, create the default config + if _, err := os.Stat(cm.configPath); os.IsNotExist(err) { + defaultConfig := &Config{ + APIServerInfo: make([]APIServerInfo, 0), + } + // save the default config + if err := cm.SaveConfig(defaultConfig); err != nil { + return nil, fmt.Errorf("failed to create the default config: [%v]", err) + } + return defaultConfig, nil + } + + // read the config file + data, err := os.ReadFile(cm.configPath) + if err != nil { + return nil, fmt.Errorf("failed to read the config file: [%v]", err) + } + + // parse the config file + var config Config + if err := json.Unmarshal(data, &config); err != nil { + return nil, fmt.Errorf("failed to parse the config file: [%s,%v]", cm.configPath, err) + } + + return &config, nil +} + +// Save is used to save the config +func (c *Config) Save() error { + usr, err := user.Current() + if err != nil { + return err + } + + // get the config directory + configDir := path.Join(usr.HomeDir, ".xce") + configPath := path.Join(configDir, "config.json") + + // create the config directory if it does not exist + if err := os.MkdirAll(configDir, 0700); err != nil { + return err + } + + // marshal the config + data, err := json.MarshalIndent(c, "", " ") + if err != nil { + return err + } + + // write the config to the file + return os.WriteFile(configPath, data, 0600) +} + +// IsAuthenticated check if the user has been authenticated by the API server +// if not, return an error +func (cm *ConfigManager) IsAuthenticated() (*APIServerInfo, error) { + cfg, err := cm.GetConfig() + if err != nil { + return nil, fmt.Errorf("failed to get config: [%v]", err) + } + + if len(cfg.APIServerInfo) == 0 { + return nil, fmt.Errorf("you are not logged in, please log in first") + } + + serverInfo := &cfg.APIServerInfo[0] + if serverInfo.URL == "" { + return nil, fmt.Errorf("no default API Server is set, please set a default server") + } + + if serverInfo.Token == "" { + return nil, fmt.Errorf("you are not logged in, please log in first") + } + + // Use the token expire time from server configuration + expireTime := serverInfo.TokenExpireTime + if expireTime == 0 { + expireTime = 86400 // Default to 24 hours if not set + } + if time.Now().Unix()-serverInfo.CreateTime > expireTime { + return nil, fmt.Errorf("your login has expired, please log in again") + } + + return serverInfo, nil +} + +// GetServerInfo +func (cm *ConfigManager) GetServerInfo() (*APIServerInfo, error) { + cfg, err := cm.GetConfig() + if err != nil { + return nil, fmt.Errorf("failed to get config: [%v]", err) + } + + if len(cfg.APIServerInfo) == 0 || cfg.APIServerInfo[0].URL == "" { + return nil, fmt.Errorf("not found the API Server information") + } + + return &cfg.APIServerInfo[0], nil +} // NewConfigManagerWithPath creates a ConfigManager using a specific config file path. // Used in tests. diff --git a/src/frontend/xcecli/models/logon.go b/src/frontend/xcecli/models/logon.go index e1fba09e..a78fc766 100755 --- a/src/frontend/xcecli/models/logon.go +++ b/src/frontend/xcecli/models/logon.go @@ -1,19 +1,18 @@ -package models - -// LogonData define the data structure in the logon response -type LogonData struct { - Token string `json:"token"` - Path string `json:"path"` - IP string `json:"ip"` - CreateTime int64 `json:"createtime"` - TokenExpireTime int64 `json:"token_expire_time"` -} - -// LogonResponse define the data structure in the logon response -type LogonResponse struct { - Code int `json:"code"` - Msg string `json:"msg"` - Data LogonData `json:"data"` - Count int `json:"count"` - CACert string `json:"cacert"` -} +package models + +// LogonData define the data structure in the logon response +type LogonData struct { + Token string `json:"token"` + Path string `json:"path"` + IP string `json:"ip"` + CreateTime int64 `json:"createtime"` + TokenExpireTime int64 `json:"token_expire_time"` +} + +// LogonResponse define the data structure in the logon response +type LogonResponse struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data LogonData `json:"data"` + Count int `json:"count"` +} diff --git a/src/include/xce/cluster.h b/src/include/xce/cluster.h index 4e04e4ca..f6e5f12c 100644 --- a/src/include/xce/cluster.h +++ b/src/include/xce/cluster.h @@ -23,8 +23,9 @@ typedef struct { /** * Cluster connection configuration. - * Populated by each caller (worker/scheduler/apiserver) from --peers and - * --ca-cert CLI flags. libxce never reads config files. + * Populated by each caller (worker/scheduler/apiserver) from the --peers + * CLI flag. libxce never reads config files. TLS is handled server-side + * (encryption without client-side peer certificate verification). * * candidates_count must be in [1, MAX_HA_CANDIDATES]. * A WARN is logged at startup if candidates_count > 3. @@ -32,9 +33,8 @@ typedef struct { typedef struct { xce_candidate_t candidates[MAX_HA_CANDIDATES]; int candidates_count; - char ca_cert_path[4096]; /* path to CA cert for TLS verify */ - char identity[256]; /* "from" name sent to controller */ - char key[256]; /* authentication key */ + char identity[256]; /* "from" name sent to controller */ + char key[256]; /* authentication key */ } xce_cluster_config_t; /** @@ -75,7 +75,6 @@ typedef struct { * Returns 0 on success, -EINVAL if: * - candidates_count < 1 or > MAX_HA_CANDIDATES * - any candidate has empty hostname or port <= 0 - * - ca_cert_path is empty * - identity is empty */ int xce_cluster_config_validate(const xce_cluster_config_t* cfg); @@ -121,8 +120,7 @@ void xce_reconnect_fini(xce_reconnect_ctx_t* rctx); int xce_parse_peers(const char* peers_str, xce_cluster_config_t* cfg); /* Test seam: set to non-NULL to override the real connect (unit tests only) */ -typedef void* (*xce_connect_fn_t)(const char* host, int port, - const char* ca_cert); +typedef void* (*xce_connect_fn_t)(const char* host, int port); void xce_cluster_set_connect_fn(xce_connect_fn_t fn); #endif /* XCE_CLUSTER_H */ diff --git a/src/lib/cluster.c b/src/lib/cluster.c index 32544bb2..bd06dceb 100644 --- a/src/lib/cluster.c +++ b/src/lib/cluster.c @@ -70,14 +70,13 @@ void xce_cluster_set_connect_fn(xce_connect_fn_t fn) /* do_connect: calls the mock when set, otherwise returns NULL (stub for * the real SSL connect via netio — to be wired in a later task). */ -static void* do_connect(const char* host, int port, const char* ca_cert) +static void* do_connect(const char* host, int port) { if (s_connect_fn) - return s_connect_fn(host, port, ca_cert); + return s_connect_fn(host, port); /* Production path: real SSL connect via netio — stub for now */ (void)host; (void)port; - (void)ca_cert; return NULL; } @@ -95,8 +94,6 @@ int xce_cluster_config_validate(const xce_cluster_config_t* cfg) if (cfg->candidates[i].port <= 0) return -EINVAL; } - if (cfg->ca_cert_path[0] == '\0') - return -EINVAL; if (cfg->identity[0] == '\0') return -EINVAL; if (cfg->candidates_count > 3) @@ -115,8 +112,7 @@ void* xce_connect_cluster(xce_cluster_config_t* cfg) for (int i = 0; i < cfg->candidates_count; i++) { void* ep = do_connect(cfg->candidates[i].hostname, - cfg->candidates[i].port, - cfg->ca_cert_path); + cfg->candidates[i].port); if (ep) return ep; } diff --git a/src/lib/testsuite/test_cluster.c b/src/lib/testsuite/test_cluster.c index 7e4a398e..c5626097 100644 --- a/src/lib/testsuite/test_cluster.c +++ b/src/lib/testsuite/test_cluster.c @@ -11,9 +11,8 @@ static int s_connect_fail_n = 0; /* fail first N attempts, then succeed */ static char s_last_host[256] = ""; static int s_last_port = 0; -static void* mock_connect(const char* host, int port, const char* ca_cert) +static void* mock_connect(const char* host, int port) { - (void)ca_cert; strncpy(s_last_host, host, sizeof(s_last_host) - 1); s_last_port = port; s_connect_call_count++; @@ -32,8 +31,6 @@ static xce_cluster_config_t make_valid_config(void) XCE_MAXHOSTNAME_LEN - 1); cfg.candidates[1].port = 7001; cfg.candidates_count = 2; - strncpy(cfg.ca_cert_path, "/etc/xce/ca.pem", - sizeof(cfg.ca_cert_path) - 1); strncpy(cfg.identity, "worker-01", sizeof(cfg.identity) - 1); return cfg; } diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 929bec3d..13ab9e08 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -122,8 +122,7 @@ typedef struct { } xce_scheduler_t; extern int sched_init(xce_scheduler_t*, xce_client_config_t*, - xce_request_handler_t); -extern void sched_set_cluster_cfg(xce_scheduler_t*, const xce_cluster_config_t*); + const xce_cluster_config_t*, xce_request_handler_t); extern void sched_fini(xce_scheduler_t*); extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*); extern int sched_update(xce_scheduler_t*); diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index 5ea45ddf..dcb51322 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -27,15 +27,13 @@ const char* argp_program_bug_address = BUG_EMAIL; struct arguments { char* logfile; int verbose; - char* peers; /* --peers value (optional) */ - char* ca_cert; /* --ca-cert value (optional) */ + char* peers; /* --peers value (optional) */ }; static struct argp_option options[] = { - { "logfile", 'l', "FILE", 0, "The full path of log file" }, - { "verbose", 'v', 0, 0, "Produce verbose log" }, - { "peers", 'P', "LIST", 0, "HA candidate list: [name=]host:port,..." }, - { "ca-cert", 'C', "FILE", 0, "CA certificate path for TLS verification" }, + { "logfile", 'l', "FILE", 0, "The full path of log file" }, + { "verbose", 'v', 0, 0, "Produce verbose log" }, + { "peers", 'P', "LIST", 0, "HA candidate list: [name=]host:port,..." }, { 0 } }; @@ -53,9 +51,6 @@ static error_t parse_opt(int key, char* arg, struct argp_state* state) case 'P': arguments->peers = arg; break; - case 'C': - arguments->ca_cert = arg; - break; default: return ARGP_ERR_UNKNOWN; } @@ -106,7 +101,9 @@ int main(int argc, char* argv[]) char* loglevel = "info"; int logsize = 64 * 1024 * 1024; int logcount = 10; - xce_client_config_t config; + xce_client_config_t config; + xce_cluster_config_t cluster_cfg = {0}; + const xce_cluster_config_t* cluster_cfg_ptr = NULL; memset(&arguments, 0x0, sizeof(arguments)); @@ -124,8 +121,6 @@ int main(int argc, char* argv[]) /* P2.5-6: populate and validate xce_cluster_config_t when --peers given */ if (arguments.peers) { - xce_cluster_config_t cluster_cfg = {0}; - if (xce_parse_peers(arguments.peers, &cluster_cfg) != 0) { fprintf(stderr, "xce_scheduler: --peers parse error: expected " @@ -133,16 +128,12 @@ int main(int argc, char* argv[]) arguments.peers); return 1; } - if (arguments.ca_cert) { - strncpy(cluster_cfg.ca_cert_path, arguments.ca_cert, - sizeof(cluster_cfg.ca_cert_path) - 1); - } strncpy(cluster_cfg.identity, "xce_scheduler", sizeof(cluster_cfg.identity) - 1); if (xce_cluster_config_validate(&cluster_cfg) != 0) { fprintf(stderr, "xce_scheduler: invalid cluster config (missing " - "--ca-cert?).\n"); + "identity?).\n"); return 1; } if (cluster_cfg.candidates_count > 3) { @@ -150,17 +141,16 @@ int main(int argc, char* argv[]) "only <=3 validated for HA", cluster_cfg.candidates_count); } - INFO("HA cluster config: %d candidate(s), identity=%s, ca_cert=%s", - cluster_cfg.candidates_count, cluster_cfg.identity, - cluster_cfg.ca_cert_path); - sched_set_cluster_cfg(&sched, &cluster_cfg); + INFO("HA cluster config: %d candidate(s), identity=%s", + cluster_cfg.candidates_count, cluster_cfg.identity); + cluster_cfg_ptr = &cluster_cfg; } else { INFO("No --peers flag; using single-controller path"); } init_config(&config); - sched_init(&sched, &config, scheduler_req_handler); + sched_init(&sched, &config, cluster_cfg_ptr, scheduler_req_handler); signal(SIGINT, sig_handler); diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 19467942..395ca969 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -372,6 +372,7 @@ static int sched_plugin_init(xce_scheduler_t* sched) } int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, + const xce_cluster_config_t* cluster_cfg, xce_request_handler_t handler) { int rc = 0; @@ -412,6 +413,10 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, memset(sched, 0x0, sizeof(*sched)); + if (cluster_cfg != NULL) { + sched->cluster_cfg = *cluster_cfg; + } + resreq_init(1); /* enable CSE intern table for scheduler process */ sched->nextid = 1; @@ -461,10 +466,22 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, /* HA mode: iterate candidates until one accepts as primary */ if (sched->cluster_cfg.candidates_count > 0) { int joined = 0; + + /* loop below re-inits apictx per candidate; free the ctx from the + * unconditional init above before the first iteration replaces it */ + xcecli_ctx_fini(sched->apictx); + for (int i = 0; i < sched->cluster_cfg.candidates_count; i++) { const char* host = sched->cluster_cfg.candidates[i].hostname; int port = sched->cluster_cfg.candidates[i].port; + /* xcecli_ctx_init() stores this pointer (ctx->conf), not a copy — + * client_failover.c dereferences it for the ctx's entire life + * (runtime reconnect), well past this loop returning. It MUST + * point at storage that outlives sched_init(): mutate the + * caller's config in place (main.c's `config` lives for the + * whole process) rather than a per-iteration stack local, which + * would leave ctx->conf dangling the moment this loop exits. */ config->nmasters = 1; config->masters[0] = (char*)host; config->master_ports[0] = port; @@ -518,15 +535,6 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, return rc; } -void sched_set_cluster_cfg(xce_scheduler_t* sched, - const xce_cluster_config_t* cfg) -{ - if (sched == NULL || cfg == NULL) { - return; - } - sched->cluster_cfg = *cfg; -} - void sched_fini(xce_scheduler_t* sched) { void* res; diff --git a/src/worker/src/main.c b/src/worker/src/main.c index f4fa180a..6ea6cc98 100644 --- a/src/worker/src/main.c +++ b/src/worker/src/main.c @@ -289,8 +289,7 @@ static int check_user(void) } /* Long-option identifiers (beyond ASCII range to avoid collisions) */ -#define OPT_PEERS 256 -#define OPT_CA_CERT 257 +#define OPT_PEERS 256 int main(int argc, char* argv[]) { @@ -301,7 +300,6 @@ int main(int argc, char* argv[]) char* custom_config_path = NULL; __attribute__((unused)) char worker_log[1024] = { 0 }; const char* peers_str = NULL; - const char* ca_cert_str = NULL; xce_cluster_config_t cluster_cfg = {0}; int use_ha = 0; /* 1 when --peers given */ @@ -394,10 +392,9 @@ int main(int argc, char* argv[]) } static struct option long_opts[] = { - { "peers", required_argument, NULL, OPT_PEERS }, - { "ca-cert", required_argument, NULL, OPT_CA_CERT }, - { "help", no_argument, NULL, 'h' }, - { NULL, 0, NULL, 0 } + { "peers", required_argument, NULL, OPT_PEERS }, + { "help", no_argument, NULL, 'h' }, + { NULL, 0, NULL, 0 } }; while ((opt = getopt_long(argc, argv, "c:p:m:n:l:h", long_opts, NULL)) @@ -440,19 +437,16 @@ int main(int argc, char* argv[]) case OPT_PEERS: peers_str = optarg; break; - case OPT_CA_CERT: - ca_cert_str = optarg; - break; case 'h': default: #ifdef VIRTUAL_CLUSTER printf("Usage: %s -n virtual_name [-p remote_port] [-m " "remote_host] [-l log_dir] [--peers list] " - "[--ca-cert path] [-h help]\n", + "[-h help]\n", argv[0]); #else printf("Usage: %s [-p remote_port] [-m remote_host] [-l log_dir] " - "[--peers ctrl-1=host:port,...] [--ca-cert path] " + "[--peers ctrl-1=host:port,...] " "[-h help]\n", argv[0]); #endif @@ -469,16 +463,12 @@ int main(int argc, char* argv[]) peers_str); return 1; } - if (ca_cert_str) { - strncpy(cluster_cfg.ca_cert_path, ca_cert_str, - sizeof(cluster_cfg.ca_cert_path) - 1); - } strncpy(cluster_cfg.identity, config.virutal_name, sizeof(cluster_cfg.identity) - 1); if (xce_cluster_config_validate(&cluster_cfg) != 0) { fprintf(stderr, - "xcecd: invalid cluster config (missing ca-cert or " - "identity?). Use --ca-cert and -n.\n"); + "xcecd: invalid cluster config (missing identity?). " + "Use -n.\n"); return 1; } if (cluster_cfg.candidates_count > 3) { @@ -488,10 +478,8 @@ int main(int argc, char* argv[]) cluster_cfg.candidates_count); } XCECD_LOG(XCE_LOG_INFO, - "HA cluster config: %d candidate(s), identity=%s, " - "ca_cert=%s", - cluster_cfg.candidates_count, cluster_cfg.identity, - cluster_cfg.ca_cert_path); + "HA cluster config: %d candidate(s), identity=%s", + cluster_cfg.candidates_count, cluster_cfg.identity); use_ha = 1; } else { XCECD_LOG(XCE_LOG_INFO,