lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(scheduler,cli): bug-1453/1454/1455/1456 remove dead CA-cert plumbing, fix scheduler HA config bugs

Browse source 
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.
This commit is contained in:
Rongsong Shen 2026-07-20 12:01:10 +00:00
parent 619f41f70e
commit 94f50ce6e1
34 changed files with 1999 additions and 2274 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/controller/main.c
View file

@ -275,10 +275,6 @@ int main(int argc, char* argv[])
return 1; 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, strncpy(myconf.peers_cfg.identity, arguments.myname,
sizeof(myconf.peers_cfg.identity) - 1); sizeof(myconf.peers_cfg.identity) - 1);
INFO("HA mode: %d controller candidates, identity=%s", INFO("HA mode: %d controller candidates, identity=%s",

82
src/frontend/xcecli/cmd/auth/cert/cert.go
View file

@ -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
}

37
src/frontend/xcecli/cmd/auth/cert/parser.go
View file

@ -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
}

360
src/frontend/xcecli/cmd/auth/logon.go
View file

@ -1,189 +1,171 @@
package cmd package cmd
import ( import (
"fmt" "fmt"
"net/url" "os"
"os" "strings"
"strings" "syscall"
"syscall" "xcecli/common/request"
"xcecli/cmd/auth/cert" "xcecli/config"
"xcecli/common/request" logger "xcecli/config"
"xcecli/config" "xcecli/models"
logger "xcecli/config"
"xcecli/models" "go.uber.org/zap"
"golang.org/x/term"
"go.uber.org/zap"
"golang.org/x/term" "github.com/spf13/cobra"
)
"github.com/spf13/cobra"
) // updateServerConfig
func updateServerConfig(cfg *config.Config, url string, loginResp *models.LogonResponse) error {
// updateServerConfig logger.Debug("updateServerConfig", zap.String("url", url))
func updateServerConfig(cfg *config.Config, url, certPath string, loginResp *models.LogonResponse) error { logger.Info("updateServerConfig", zap.String("url", url))
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
// check if the server with the same URL already exists for i, server := range cfg.APIServerInfo {
Found := false if server.URL == url {
for i, server := range cfg.APIServerInfo { // update the existing server info
if server.URL == url { cfg.APIServerInfo[i].Token = loginResp.Data.Token
// update the existing server info cfg.APIServerInfo[i].Path = loginResp.Data.Path //save the path
cfg.APIServerInfo[i].Token = loginResp.Data.Token cfg.APIServerInfo[i].CreateTime = loginResp.Data.CreateTime
cfg.APIServerInfo[i].Path = loginResp.Data.Path //save the path Found = true
cfg.APIServerInfo[i].CreateTime = loginResp.Data.CreateTime break
cfg.APIServerInfo[i].CACert = certPath }
Found = true }
break
} // if it's a new server, add it to the list
} if !Found {
// if it's a new server, add it to the list newServer := config.APIServerInfo{
if !Found { URL: url,
Token: loginResp.Data.Token,
newServer := config.APIServerInfo{ Path: loginResp.Data.Path,
URL: url, CreateTime: loginResp.Data.CreateTime,
Token: loginResp.Data.Token, }
Path: loginResp.Data.Path,
CreateTime: loginResp.Data.CreateTime, // add it to the end of the array
CACert: certPath, cfg.APIServerInfo = append(cfg.APIServerInfo, newServer)
} logger.Debug("newServer", zap.Any("newServer", newServer))
logger.Info("newServer", zap.Any("newServer", newServer))
// add it to the end of the array }
cfg.APIServerInfo = append(cfg.APIServerInfo, newServer)
logger.Debug("newServer", zap.Any("newServer", newServer)) return nil
logger.Info("newServer", zap.Any("newServer", newServer)) }
}
func NewLogonCmd(configManager *config.ConfigManager) *cobra.Command {
return nil var opts struct {
} username string
password string
func NewLogonCmd(configManager *config.ConfigManager) *cobra.Command { url string
var opts struct { }
username string logger.Debug("NewLogonCmd", zap.Any("opts", opts))
password string logger.Info("NewLogonCmd", zap.Any("opts", opts))
url string
} cmd := &cobra.Command{
logger.Debug("NewLogonCmd", zap.Any("opts", opts)) Use: "login",
logger.Info("NewLogonCmd", zap.Any("opts", opts)) GroupID: "api",
Short: "Login to API Server",
cmd := &cobra.Command{ Long: `Login to the specified API Server, and save the login info to the config file.
Use: "login", Example:
GroupID: "api", xce login -u USERNAME -p PASSWORD --url http://IP:8080
Short: "Login to API Server", xce login -u USERNAME -p PASSWORD --url https://IP:8443`,
Long: `Login to the specified API Server, and save the login info to the config file. Run: func(cmd *cobra.Command, args []string) {
Example: // if the password is not provided, prompt the user to enter it
xce login -u USERNAME -p PASSWORD --url http://IP:8080 if opts.password == "" {
xce login -u USERNAME -p PASSWORD --url https://IP:8443`, fmt.Print("enter password: ")
Run: func(cmd *cobra.Command, args []string) { bytePassword, err := term.ReadPassword(int(syscall.Stdin))
// if the password is not provided, prompt the user to enter it fmt.Println() // new line
if opts.password == "" { if err != nil {
fmt.Print("enter password: ") errMsg := request.ExtractErrorMessage(err)
bytePassword, err := term.ReadPassword(int(syscall.Stdin)) fmt.Printf("error reading password: [%s]", errMsg)
fmt.Println() // new line os.Exit(1)
if err != nil { }
errMsg := request.ExtractErrorMessage(err) opts.password = string(bytePassword)
fmt.Printf("error reading password: [%s]", errMsg) }
os.Exit(1)
} err := runLogon(opts, configManager)
opts.password = string(bytePassword) if err != nil {
} fmt.Println(strings.ToLower(err.Error()))
}
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)")
// set the command line parameters flags.StringVar(&opts.url, "url", "", "API Server URL")
flags := cmd.Flags()
flags.StringVarP(&opts.username, "username", "u", "", "username") // required parameters
flags.StringVarP(&opts.password, "password", "p", "", "password (optional, will prompt if not provided)") cmd.MarkFlagRequired("username")
flags.StringVar(&opts.url, "url", "", "API Server URL")
//cmd.MarkFlagRequired("password")
// required parameters cmd.MarkFlagRequired("url")
cmd.MarkFlagRequired("username")
logger.Debug("cmd", zap.Any("cmd", cmd))
//cmd.MarkFlagRequired("password") logger.Info("cmd", zap.Any("cmd", cmd))
cmd.MarkFlagRequired("url")
return cmd
logger.Debug("cmd", zap.Any("cmd", cmd)) }
logger.Info("cmd", zap.Any("cmd", cmd))
func runLogon(opts struct{ username, password, url string }, cm *config.ConfigManager) error {
return cmd if cm == nil {
} return fmt.Errorf("fail to initialize the config manager")
}
func runLogon(opts struct{ username, password, url string }, cm *config.ConfigManager) error { logger.Debug("runLogon", zap.Any("opts", opts))
if cm == nil { logger.Info("runLogon", zap.Any("opts", opts))
return fmt.Errorf("fail to initialize the config manager")
} //get the config
logger.Debug("runLogon", zap.Any("opts", opts)) cfg, err := cm.GetConfig()
logger.Info("runLogon", zap.Any("opts", opts)) if err != nil {
cfg = &config.Config{}
//get the config }
cfg, err := cm.GetConfig() logger.Debug("cfg", zap.Any("cfg", cfg))
if err != nil { logger.Info("cfg", zap.Any("cfg", cfg))
cfg = &config.Config{}
} // create the API client
logger.Debug("cfg", zap.Any("cfg", cfg)) apiClient := request.NewAPIClient(opts.url)
logger.Info("cfg", zap.Any("cfg", cfg)) logger.Debug("apiClient", zap.Any("apiClient", apiClient))
logger.Info("apiClient", zap.Any("apiClient", apiClient))
// create the API client
apiClient := request.NewAPIClient(opts.url) // perform the login
certPath := "" loginResp, err := apiClient.Logon(opts.username, opts.password)
// Only setup certs for HTTPS connections if err != nil {
if u, err := url.Parse(opts.url); err == nil && u.Scheme == "https" { // If login fails, remove the server entry with the same URL if it exists
certPath, err = cert.GetCertPath() for i, server := range cfg.APIServerInfo {
apiClient.SetRootCAs(certPath) if server.URL == opts.url {
if err != nil { // Remove this server from the slice
errMsg := request.ExtractErrorMessage(err) cfg.APIServerInfo = append(cfg.APIServerInfo[:i], cfg.APIServerInfo[i+1:]...)
return fmt.Errorf("failed to get cert path: [%s]", errMsg) // Save the updated config
} if saveErr := cm.SaveConfig(cfg); saveErr != nil {
if err := cert.GeneratorCert(opts.url, certPath); err != nil { logger.Error("Failed to save the config after removing server entry", zap.Error(saveErr))
errMsg := request.ExtractErrorMessage(err) }
return fmt.Errorf("failed to generate cert: [%s]", errMsg) break
} }
} }
logger.Debug("apiClient", zap.Any("apiClient", apiClient)) errMsg := request.ExtractErrorMessage(err)
logger.Info("apiClient", zap.Any("apiClient", apiClient)) return fmt.Errorf("failed to login: [%s]", errMsg)
}
// perform the login logger.Debug("loginResp", zap.Any("loginResp", loginResp))
loginResp, err := apiClient.Logon(opts.username, opts.password) logger.Info("loginResp", zap.Any("loginResp", loginResp))
if err != nil {
// If login fails, remove the server entry with the same URL if it exists // update the server config
for i, server := range cfg.APIServerInfo { if err := updateServerConfig(cfg, opts.url, loginResp); err != nil {
if server.URL == opts.url { errMsg := request.ExtractErrorMessage(err)
// Remove this server from the slice return fmt.Errorf("failed to update the server config: [%s]", errMsg)
cfg.APIServerInfo = append(cfg.APIServerInfo[:i], cfg.APIServerInfo[i+1:]...) }
// Save the updated config logger.Debug("cfg", zap.Any("cfg", cfg))
if saveErr := cm.SaveConfig(cfg); saveErr != nil { logger.Info("cfg", zap.Any("cfg", cfg))
logger.Error("Failed to save the config after removing server entry", zap.Error(saveErr))
} // save the config
break if err := cm.SaveConfig(cfg); err != nil {
} errMsg := request.ExtractErrorMessage(err)
} return fmt.Errorf("failed to save the config: [%s]", errMsg)
errMsg := request.ExtractErrorMessage(err) }
return fmt.Errorf("failed to login: [%s]", errMsg) logger.Debug("cfg", zap.Any("cfg", cfg))
} logger.Info("cfg", zap.Any("cfg", cfg))
logger.Debug("loginResp", zap.Any("loginResp", loginResp))
logger.Info("loginResp", zap.Any("loginResp", loginResp)) fmt.Println("login successfully")
return nil
// 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
}

159
src/frontend/xcecli/cmd/clusters/list.go
View file

@ -1,80 +1,79 @@
package clusters package clusters
import ( import (
"fmt" "fmt"
"strings" "strings"
"xcecli/common/output" "xcecli/common/output"
"xcecli/common/request" "xcecli/common/request"
"xcecli/config" "xcecli/config"
"xcecli/models" "xcecli/models"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
var ( var (
cluster string cluster string
fields string fields string
Verbose bool Verbose bool
) )
// NewClustersCmd create cluster query command // NewClustersCmd create cluster query command
func NewClustersCmd(configManager *config.ConfigManager) *cobra.Command { func NewClustersCmd(configManager *config.ConfigManager) *cobra.Command {
cmd := &cobra.Command{ cmd := &cobra.Command{
Use: "id", Use: "id",
GroupID: "query", GroupID: "query",
Short: "Query cluster info", Short: "Query cluster info",
Long: `Query cluster info, support filter by cluster name . Long: `Query cluster info, support filter by cluster name .
example: example:
xce clusters # query all clusters`, xce clusters # query all clusters`,
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
err := runClusters(configManager) err := runClusters(configManager)
if err != nil { if err != nil {
fmt.Println(strings.ToLower(err.Error())) fmt.Println(strings.ToLower(err.Error()))
return nil return nil
} }
return nil return nil
}, },
} }
//cmd.Flags().BoolVarP(&Verbose, "verbose", "l", false, "show cluster detail") //cmd.Flags().BoolVarP(&Verbose, "verbose", "l", false, "show cluster detail")
return cmd return cmd
} }
func runClusters(cm *config.ConfigManager) error { func runClusters(cm *config.ConfigManager) error {
// get the config // get the config
serverInfo, _ := cm.GetServerInfo() serverInfo, _ := cm.GetServerInfo()
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// validate the parameters
// validate the parameters cluster = strings.ToLower(cluster)
cluster = strings.ToLower(cluster)
// create the query parameters
// create the query parameters params := make(map[string]string)
params := make(map[string]string)
// handle field selection
// handle field selection if fields != "" || cluster != "" {
if fields != "" || cluster != "" { params["fields"] = fields
params["fields"] = fields }
}
// create the API client and query the host info
// create the API client and query the host info var resp models.ClusterResponse
var resp models.ClusterResponse err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/xce/v1/clusters", &resp, params)
err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, "/xce/v1/clusters", &resp, params) if err != nil {
if err != nil { errMsg := request.ExtractErrorMessage(err)
errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to get clusters: [%s]", errMsg)
return fmt.Errorf("failed to get clusters: [%s]", errMsg) }
}
if resp.Data.Count == 0 {
if resp.Data.Count == 0 { fmt.Println("no cluster found")
fmt.Println("no cluster found") return nil
return nil }
}
// print result
// print result output := output.NewOutput(fields, false, false)
output := output.NewOutput(fields, false, false) output.SetVerbose(Verbose)
output.SetVerbose(Verbose) output.ClusterPrintData(&resp.Data.Rows[0])
output.ClusterPrintData(&resp.Data.Rows[0]) return nil
return nil }
}

356
src/frontend/xcecli/cmd/hosts/controller.go
View file

@ -1,179 +1,177 @@
package hosts package hosts
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"strings" "strings"
"xcecli/common/request" "xcecli/common/request"
"xcecli/config" "xcecli/config"
"xcecli/models" "xcecli/models"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
const ( const (
HOST_OPEN = 1 HOST_OPEN = 1
HOST_CLOSE = 2 HOST_CLOSE = 2
) )
// HostController host control controller // HostController host control controller
type HostController struct { type HostController struct {
configManager *config.ConfigManager configManager *config.ConfigManager
action string action string
hostnames []string hostnames []string
comment string comment string
allHosts bool allHosts bool
} }
// NewHostController create host control controller // NewHostController create host control controller
func NewHostController(configManager *config.ConfigManager, action string) *HostController { func NewHostController(configManager *config.ConfigManager, action string) *HostController {
return &HostController{ return &HostController{
configManager: configManager, configManager: configManager,
action: action, action: action,
hostnames: []string{}, hostnames: []string{},
} }
} }
// Run execute host control // Run execute host control
func (c *HostController) RunHostsControl() error { func (c *HostController) RunHostsControl() error {
// get server info and verify authentication // get server info and verify authentication
serverInfo, _ := c.configManager.GetServerInfo() serverInfo, _ := c.configManager.GetServerInfo()
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// handle all hosts case
// handle all hosts case if c.allHosts {
if c.allHosts { c.hostnames = []string{"all"}
c.hostnames = []string{"all"} }
}
// if no hostnames specified, return error
// if no hostnames specified, return error if len(c.hostnames) == 0 {
if len(c.hostnames) == 0 { return fmt.Errorf("must specify hostnames or use --all option")
return fmt.Errorf("must specify hostnames or use --all option") }
}
// create request
// create request req := models.HostControlRequest{
req := models.HostControlRequest{ ControlAction: c.action,
ControlAction: c.action, Hostname: c.hostnames,
Hostname: c.hostnames, Comment: c.comment,
Comment: c.comment, }
}
// encode request to JSON bytes
// encode request to JSON bytes reqData, err := json.Marshal(req)
reqData, err := json.Marshal(req) if err != nil {
if err != nil { errMsg := request.ExtractErrorMessage(err)
errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to encode request data: [%s]", errMsg)
return fmt.Errorf("failed to encode request data: [%s]", errMsg) }
}
var resp models.HostControlResponse
var resp models.HostControlResponse err = apiClient.PostRequestToAPIServer(
err = apiClient.PostRequestToAPIServer( serverInfo.Token,
serverInfo.Token, "/xce/v1/hosts/control",
serverInfo.CACert, reqData,
"/xce/v1/hosts/control", &resp,
reqData, )
&resp,
) if err != nil {
return err
if err != nil { }
return err
} if len(c.hostnames) == 1 && c.hostnames[0] == "all" {
for _, data := range resp.Data.Accepted {
if len(c.hostnames) == 1 && c.hostnames[0] == "all" { fmt.Printf("host <%s> is being <%s>\n", data.HostName, c.action)
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)
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 {
} else { fmt.Printf("host <%s> is being <%s>\n", data.HostName, c.action)
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)
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
} }
return nil
} // Parse parse command line arguments
func (c *HostController) ParseCmdArgs(cmd *cobra.Command, args []string) error {
// Parse parse command line arguments // handle hostnames
func (c *HostController) ParseCmdArgs(cmd *cobra.Command, args []string) error { if len(args) > 0 {
// handle hostnames c.hostnames = append(c.hostnames, args...)
if len(args) > 0 { }
c.hostnames = append(c.hostnames, args...)
} return nil
}
return nil
} // NewOpenHostsCmd create open hosts command
func NewOpenHostsCmd(configManager *config.ConfigManager) *cobra.Command {
// NewOpenHostsCmd create open hosts command
func NewOpenHostsCmd(configManager *config.ConfigManager) *cobra.Command { controller := NewHostController(configManager, "open")
controller := NewHostController(configManager, "open") cmd := &cobra.Command{
Use: "open [hostname1 hostname2...]",
cmd := &cobra.Command{ Short: "Open hosts",
Use: "open [hostname1 hostname2...]", Long: `Open hosts by hostname.
Short: "Open hosts", Open hosts changes hosts from DOWN, UNREACH, or UNAVAIL status to OK status.
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
Examples: xce hosts open --all # open all hosts`,
xce hosts open host1,host2,... # open specific hosts RunE: func(cmd *cobra.Command, args []string) error {
xce hosts open --all # open all hosts`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunHostsControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunHostsControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.BoolVarP(&controller.allHosts, "all", "a", false, "apply to all hosts")
flags := cmd.Flags() flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation")
flags.BoolVarP(&controller.allHosts, "all", "a", false, "apply to all hosts")
flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") return cmd
}
return cmd
} // NewCloseHostsCmd create close hosts command
func NewCloseHostsCmd(configManager *config.ConfigManager) *cobra.Command {
// NewCloseHostsCmd create close hosts command controller := NewHostController(configManager, "close")
func NewCloseHostsCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewHostController(configManager, "close") cmd := &cobra.Command{
Use: "close [hostname1 hostname2...]",
cmd := &cobra.Command{ Short: "Close hosts",
Use: "close [hostname1 hostname2...]", Long: `Close hosts by hostname.
Short: "Close hosts", Close hosts changes hosts from OK, BUSY, or IDLE status to UNAVAIL status.
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
Examples: xce hosts close --all # close all hosts`,
xce hosts close host1 host2 # close specific hosts RunE: func(cmd *cobra.Command, args []string) error {
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()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunHostsControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunHostsControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.BoolVarP(&controller.allHosts, "all", "a", false, "apply to all hosts")
flags := cmd.Flags() flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation")
flags.BoolVarP(&controller.allHosts, "all", "a", false, "apply to all hosts")
flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") return cmd
}
return cmd
}

397
src/frontend/xcecli/cmd/hosts/list.go
View file

@ -1,199 +1,198 @@
package hosts package hosts
import ( import (
"fmt" "fmt"
"strings" "strings"
"xcecli/common/output" "xcecli/common/output"
"xcecli/common/request" "xcecli/common/request"
"xcecli/config" "xcecli/config"
logger "xcecli/config" logger "xcecli/config"
host_filter "xcecli/filters" host_filter "xcecli/filters"
"xcecli/models" "xcecli/models"
"go.uber.org/zap" "go.uber.org/zap"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
// HostsQuerier struct for querying hosts // HostsQuerier struct for querying hosts
type HostsQuerier struct { type HostsQuerier struct {
configManager *config.ConfigManager configManager *config.ConfigManager
filter *host_filter.HostsFilter filter *host_filter.HostsFilter
fields string fields string
jsonOutput bool jsonOutput bool
noHeader bool noHeader bool
verbose bool verbose bool
wideFormat bool // wide format wideFormat bool // wide format
} }
// NewHostsQuerier creates a new hosts querier // NewHostsQuerier creates a new hosts querier
func NewHostsQuerier(configManager *config.ConfigManager) *HostsQuerier { func NewHostsQuerier(configManager *config.ConfigManager) *HostsQuerier {
return &HostsQuerier{ return &HostsQuerier{
configManager: configManager, configManager: configManager,
filter: &host_filter.HostsFilter{}, filter: &host_filter.HostsFilter{},
} }
} }
// Parse parses command line arguments, simplified version that doesn't need to process flags already bound to filter // 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 { func (hq *HostsQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error {
// No additional processing needed, all parameters are bound in flags // No additional processing needed, all parameters are bound in flags
if len(args) > 0 { if len(args) > 0 {
hq.filter.HostName = strings.Join(args, ",") hq.filter.HostName = strings.Join(args, ",")
} }
// Define mutual exclusions for verbose flag // Define mutual exclusions for verbose flag
if hq.verbose { if hq.verbose {
conflicts := make([]string, 0) conflicts := make([]string, 0)
if hq.wideFormat { if hq.wideFormat {
conflicts = append(conflicts, "--wided (-w)") conflicts = append(conflicts, "--wided (-w)")
} }
if hq.fields != "" { if hq.fields != "" {
conflicts = append(conflicts, "--output (-o)") conflicts = append(conflicts, "--output (-o)")
} }
if len(conflicts) > 0 { if len(conflicts) > 0 {
return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s",
strings.Join(conflicts, ", "), cmd.UsageString()) strings.Join(conflicts, ", "), cmd.UsageString())
} }
} }
return nil return nil
} }
// Run executes the hosts query command // Run executes the hosts query command
func (hq *HostsQuerier) RunHostsList() error { func (hq *HostsQuerier) RunHostsList() error {
// Get configuration // Get configuration
// Get server information // Get server information
serverInfo, _ := hq.configManager.GetServerInfo() serverInfo, _ := hq.configManager.GetServerInfo()
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Create query parameters
// Create query parameters params := make(map[string]string)
params := make(map[string]string)
// Get filter string
// Get filter string filterStr, err := hq.filter.HandleHostFilter()
filterStr, err := hq.filter.HandleHostFilter()
if err != nil {
if err != nil { errMsg := request.ExtractErrorMessage(err)
errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to get filter: [%s]", errMsg)
return fmt.Errorf("failed to get filter: [%s]", errMsg) } else if filterStr != "" {
} else if filterStr != "" { params["filter"] = filterStr
params["filter"] = filterStr }
}
// Field mapping
// Field mapping fieldsMap := models.HostFieldsMap
fieldsMap := models.HostFieldsMap
if hq.verbose && hq.fields == "" {
if hq.verbose && hq.fields == "" { for header := range fieldsMap {
for header := range fieldsMap { hq.fields += header + ","
hq.fields += header + "," }
} hq.fields = strings.TrimSuffix(hq.fields, ",")
hq.fields = strings.TrimSuffix(hq.fields, ",") }
} // Process field selection
// Process field selection if hq.fields != "" {
if hq.fields != "" { hq.fields = request.CleanFields(hq.fields)
hq.fields = request.CleanFields(hq.fields) mappedFields, err := request.CheckAndMapFields(hq.fields, fieldsMap)
mappedFields, err := request.CheckAndMapFields(hq.fields, fieldsMap) if err != nil {
if err != nil { return err
return err }
} params["fields"] = strings.Join(mappedFields, ",")
params["fields"] = strings.Join(mappedFields, ",") }
}
// create page printer
// create page printer pagePrinter := output.NewPagePrinter(output.DataTypeHosts, output.WithFields(hq.fields),
pagePrinter := output.NewPagePrinter(output.DataTypeHosts, output.WithFields(hq.fields), output.WithFormatJson(hq.jsonOutput), output.WithNoHeader(hq.noHeader),
output.WithFormatJson(hq.jsonOutput), output.WithNoHeader(hq.noHeader), output.WithVerbose(hq.verbose), output.WithWideFormat(hq.wideFormat))
output.WithVerbose(hq.verbose), output.WithWideFormat(hq.wideFormat))
// Record if the first page has been processed
// Record if the first page has been processed isFirstPage := true
isFirstPage := true
var (
var ( resp models.HostsResponse
resp models.HostsResponse loopCount int
loopCount int pageToken string
pageToken string )
)
// Loop to get all pages of data
// Loop to get all pages of data for {
for { if loopCount > 0 && pageToken == "" {
if loopCount > 0 && pageToken == "" { break
break }
}
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/hosts", &resp, params)
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/hosts", &resp, params) if err != nil {
if err != nil { return err
return err }
}
if len(resp.Data.Rows) == 0 {
if len(resp.Data.Rows) == 0 { if isFirstPage {
if isFirstPage { fmt.Println("no host found")
fmt.Println("no host found") return nil
return nil }
} }
}
// Process the first page
// Process the first page if isFirstPage {
if isFirstPage { // Start printing process
// Start printing process pagePrinter.StartPrinting()
pagePrinter.StartPrinting() // Prepare for printing, calculate column width and table header, etc.
// Prepare for printing, calculate column width and table header, etc. pagePrinter.PrepareForPrinting(&resp)
pagePrinter.PrepareForPrinting(&resp) isFirstPage = false
isFirstPage = false }
}
// Print the current page
// Print the current page pagePrinter.PrintPage(&resp)
pagePrinter.PrintPage(&resp)
// Update page token and loop counter
// Update page token and loop counter pageToken = resp.Data.Page_token
pageToken = resp.Data.Page_token loopCount++
loopCount++ }
}
// Finish printing
// Finish printing pagePrinter.Finish()
pagePrinter.Finish()
logger.Debug("hosts response completed", zap.Int("pages", loopCount))
logger.Debug("hosts response completed", zap.Int("pages", loopCount)) return nil
return nil }
}
func NewListHostsCmd(configManager *config.ConfigManager) *cobra.Command {
func NewListHostsCmd(configManager *config.ConfigManager) *cobra.Command { hostsQuerier := NewHostsQuerier(configManager)
hostsQuerier := NewHostsQuerier(configManager)
cmd := &cobra.Command{
cmd := &cobra.Command{ Use: "list",
Use: "list", Short: "Query the host info",
Short: "Query the host info", Long: `Query the host info, support basic info and detailed info.
Long: `Query the host info, support basic info and detailed info. Example:
Example: xce hosts list # query the full info
xce hosts list # query the full info xce hosts list -a # query the detailed info for all hosts
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 .." # query the specified fields xce hosts list -o "host_name status max_cpus .." --json # query the specified fields and output in json format`,
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 {
RunE: func(cmd *cobra.Command, args []string) error { if err := hostsQuerier.ParseCmdArgs(cmd, args); err != nil {
if err := hostsQuerier.ParseCmdArgs(cmd, args); err != nil { fmt.Println(strings.ToLower(err.Error()))
fmt.Println(strings.ToLower(err.Error())) return nil
return nil }
} if err := hostsQuerier.RunHostsList(); err != nil {
if err := hostsQuerier.RunHostsList(); err != nil { fmt.Println(strings.ToLower(err.Error()))
fmt.Println(strings.ToLower(err.Error())) return nil
return nil }
} return nil
return nil },
}, }
}
// Add command line parameters
// Add command line parameters flags := cmd.Flags()
flags := cmd.Flags() flags.BoolVarP(&hostsQuerier.verbose, "host_info", "a", false, "host information")
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.filter.HostType, "host_type", "t", "", "host type (X86_64 or ARM)") flags.StringVarP(&hostsQuerier.fields, "output", "o", "", "output fields for hosts")
flags.StringVarP(&hostsQuerier.fields, "output", "o", "", "output fields for hosts") flags.BoolVarP(&hostsQuerier.wideFormat, "wided", "w", false, "output in wide format")
flags.BoolVarP(&hostsQuerier.wideFormat, "wided", "w", false, "output in wide format") flags.BoolVarP(&hostsQuerier.verbose, "verbose", "l", false, "show host detail")
flags.BoolVarP(&hostsQuerier.verbose, "verbose", "l", false, "show host detail") flags.BoolVarP(&hostsQuerier.jsonOutput, "json", "", false, "output in JSON format")
flags.BoolVarP(&hostsQuerier.jsonOutput, "json", "", false, "output in JSON format") flags.BoolVarP(&hostsQuerier.noHeader, "noheader", "n", false, "output no header")
flags.BoolVarP(&hostsQuerier.noHeader, "noheader", "n", false, "output no header")
return cmd
return cmd }
}

3
src/frontend/xcecli/cmd/hosts/select.go
View file

@ -36,7 +36,6 @@ func (s *HostSelector) run() error {
return err return err
} }
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
req := models.ResReqSelectRequest{ req := models.ResReqSelectRequest{
Select: s.resReq, Select: s.resReq,
@ -46,7 +45,7 @@ func (s *HostSelector) run() error {
body, _ := json.Marshal(req) body, _ := json.Marshal(req)
var resp models.HostsResponse 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 { "/xce/v1/hosts/select", body, &resp); err != nil {
return err return err
} }

572
src/frontend/xcecli/cmd/jobs/controller.go
View file

@ -1,287 +1,285 @@
package jobs package jobs
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"strings" "strings"
"xcecli/common/format" "xcecli/common/format"
"xcecli/common/request" "xcecli/common/request"
"xcecli/config" "xcecli/config"
"xcecli/models" "xcecli/models"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
// JobController defines job control // JobController defines job control
type JobController struct { type JobController struct {
configManager *config.ConfigManager configManager *config.ConfigManager
action string action string
filter models.JobControlFilter filter models.JobControlFilter
jobIDs []string jobIDs []string
} }
// NewJobController create job controller // NewJobController create job controller
func NewJobController(configManager *config.ConfigManager, action string) *JobController { func NewJobController(configManager *config.ConfigManager, action string) *JobController {
return &JobController{ return &JobController{
configManager: configManager, configManager: configManager,
action: action, action: action,
} }
} }
// Run execute job control // Run execute job control
func (c *JobController) RunJobsControl() error { func (c *JobController) RunJobsControl() error {
// Get server information // Get server information
serverInfo, _ := c.configManager.GetServerInfo() serverInfo, _ := c.configManager.GetServerInfo()
// Create API client // Create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
if len(c.jobIDs) > 0 {
if len(c.jobIDs) > 0 { // if job IDs are specified, use job ID interface
// if job IDs are specified, use job ID interface return c.controlJobsByIDs(apiClient, serverInfo.Token)
return c.controlJobsByIDs(apiClient, serverInfo.Token, serverInfo.CACert) } else {
} else { // otherwise use filter conditions
// otherwise use filter conditions return c.controlJobsByFilter(apiClient, serverInfo.Token)
return c.controlJobsByFilter(apiClient, serverInfo.Token, serverInfo.CACert) }
} }
}
// control jobs by job IDs
// control jobs by job IDs func (c *JobController) controlJobsByIDs(apiClient *request.APIClient, token string) error {
func (c *JobController) controlJobsByIDs(apiClient *request.APIClient, token string, certPath string) error { apiIDs := make([]string, 0, len(c.jobIDs))
apiIDs := make([]string, 0, len(c.jobIDs)) for _, id := range c.jobIDs {
for _, id := range c.jobIDs { apiID, err := format.ToAPIJobID(strings.TrimSpace(id))
apiID, err := format.ToAPIJobID(strings.TrimSpace(id)) if err != nil {
if err != nil { return fmt.Errorf("invalid job ID %q: %w", id, err)
return fmt.Errorf("invalid job ID %q: %w", id, err) }
} apiIDs = append(apiIDs, apiID)
apiIDs = append(apiIDs, apiID) }
} url := fmt.Sprintf("/xce/v1/jobs/%s/control", strings.Join(apiIDs, ","))
url := fmt.Sprintf("/xce/v1/jobs/%s/control", strings.Join(apiIDs, ","))
req := models.JobControlJobIDRequest{
req := models.JobControlJobIDRequest{ Action: c.action,
Action: c.action, }
} // encode request data to JSON bytes
// encode request data to JSON bytes return c.printJobControlResponse(req, apiClient, token, url)
return c.printJobControlResponse(req, apiClient, token, certPath, url) }
}
// control jobs by filter conditions
// control jobs by filter conditions func (c *JobController) controlJobsByFilter(apiClient *request.APIClient, token string) error {
func (c *JobController) controlJobsByFilter(apiClient *request.APIClient, token string, certPath string) error { url := "/xce/v1/jobs/control"
url := "/xce/v1/jobs/control"
req := models.JobControlRequest{
req := models.JobControlRequest{ Action: c.action,
Action: c.action, Filter: c.filter,
Filter: c.filter, }
} // encode request data to JSON bytes
// encode request data to JSON bytes return c.printJobControlResponse(req, apiClient, token, url)
return c.printJobControlResponse(req, apiClient, token, certPath, url) }
}
func (c *JobController) printJobControlResponse(req interface{}, apiClient *request.APIClient, token string, url string) error {
func (c *JobController) printJobControlResponse(req interface{}, apiClient *request.APIClient, token string, certPath string, url string) error { reqData, err := json.Marshal(req)
reqData, err := json.Marshal(req) if err != nil {
if err != nil { errMsg := request.ExtractErrorMessage(err)
errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to encode request data: [%s]", errMsg)
return fmt.Errorf("failed to encode request data: [%s]", errMsg) }
} var resp models.JobControlResponse
var resp models.JobControlResponse err = apiClient.PostRequestToAPIServer(
err = apiClient.PostRequestToAPIServer( token,
token, url,
certPath, reqData,
url, &resp,
reqData, )
&resp,
) if err != nil {
return err
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.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 "))
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
} }
return nil
} // Parse parse command line arguments
// Parse command line arguments
// Parse parse command line arguments func (c *JobController) ParseCmdArgs(cmd *cobra.Command, args []string) error {
// Parse command line arguments if len(args) > 0 {
func (c *JobController) ParseCmdArgs(cmd *cobra.Command, args []string) error { if len(args) == 1 && args[0] == "0" {
if len(args) > 0 { c.filter.AllJobs = true
if len(args) == 1 && args[0] == "0" { } else {
c.filter.AllJobs = true // parse job ID parameters
} else { c.jobIDs = strings.Split(args[0], ",")
// parse job ID parameters return nil
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 {
// check if any filter conditions are set // if no job IDs or filter conditions are specified, default to processing all jobs of the current user
if c.filter.JobName == "" && c.filter.QueueName == "" && c.filter.Users == "" && !c.filter.AllJobs { return fmt.Errorf("no job IDs or filter conditions specified")
// 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
} }
return nil
} // NewKillJobsCmd create kill job command
func NewKillJobsCmd(configManager *config.ConfigManager) *cobra.Command {
// NewKillJobsCmd create kill job command controller := NewJobController(configManager, "kill")
func NewKillJobsCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewJobController(configManager, "kill") cmd := &cobra.Command{
Use: "kill",
cmd := &cobra.Command{ Short: "Kill jobs",
Use: "kill", Long: `Kill jobs by job IDs or filters.
Short: "Kill jobs", Examples:
Long: `Kill jobs by job IDs or filters. xce jobs kill jobid1,jobid2,... # kill jobs by job IDs
Examples: xce jobs kill -J jobname # kill jobs by job name
xce jobs kill jobid1,jobid2,... # kill jobs by job IDs xce jobs kill -q queue_name # kill jobs in a specific queue
xce jobs kill -J jobname # kill jobs by job name xce jobs kill -u username # kill jobs by user`,
xce jobs kill -q queue_name # kill jobs in a specific queue RunE: func(cmd *cobra.Command, args []string) error {
xce jobs kill -u username # kill jobs by user`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunJobsControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunJobsControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name")
flags := cmd.Flags() flags.StringVarP(&controller.filter.QueueName, "queue", "q", "", "filter by queue name")
flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user")
flags.StringVarP(&controller.filter.QueueName, "queue", "q", "", "filter by queue name") flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs")
flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user")
flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") return cmd
}
return cmd
} // NewStopJobsCmd create stop job command
func NewStopJobsCmd(configManager *config.ConfigManager) *cobra.Command {
// NewStopJobsCmd create stop job command controller := NewJobController(configManager, "stop")
func NewStopJobsCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewJobController(configManager, "stop") cmd := &cobra.Command{
Use: "stop",
cmd := &cobra.Command{ Short: "Stop jobs",
Use: "stop", Long: `Stop jobs by job IDs or filters.
Short: "Stop jobs", Examples:
Long: `Stop jobs by job IDs or filters. xce jobs stop jobid1,jobid2,... # stop jobs by job IDs
Examples: xce jobs stop -J jobname # stop jobs by job name
xce jobs stop jobid1,jobid2,... # stop jobs by job IDs xce jobs stop -q queue_name # stop jobs in a specific queue
xce jobs stop -J jobname # stop jobs by job name xce jobs stop -u username # stop jobs by user`,
xce jobs stop -q queue_name # stop jobs in a specific queue RunE: func(cmd *cobra.Command, args []string) error {
xce jobs stop -u username # stop jobs by user`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunJobsControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunJobsControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name")
flags := cmd.Flags() flags.StringVarP(&controller.filter.QueueName, "queue", "q", "", "filter by queue name")
flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user")
flags.StringVarP(&controller.filter.QueueName, "queue", "q", "", "filter by queue name") flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs")
flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user")
flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") return cmd
}
return cmd
} // NewResumeJobsCmd create resume job command
func NewResumeJobsCmd(configManager *config.ConfigManager) *cobra.Command {
// NewResumeJobsCmd create resume job command controller := NewJobController(configManager, "resume")
func NewResumeJobsCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewJobController(configManager, "resume") cmd := &cobra.Command{
Use: "resume",
cmd := &cobra.Command{ Short: "Resume jobs",
Use: "resume", Long: `Resume jobs by job IDs or filters.
Short: "Resume jobs", Examples:
Long: `Resume jobs by job IDs or filters. xce jobs resume jobid1,jobid2,... # resume jobs by job IDs
Examples: xce jobs resume -J jobname # resume jobs by job name
xce jobs resume jobid1,jobid2,... # resume jobs by job IDs xce jobs resume -u username # resume jobs by user`,
xce jobs resume -J jobname # resume jobs by job name RunE: func(cmd *cobra.Command, args []string) error {
xce jobs resume -u username # resume jobs by user`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunJobsControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunJobsControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name")
flags := cmd.Flags() flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user")
flags.StringVarP(&controller.filter.JobName, "job_name", "J", "", "filter by job name") flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs")
flags.StringVarP(&controller.filter.Users, "user", "u", "", "filter by user")
flags.BoolVarP(&controller.filter.AllJobs, "all", "a", false, "select all jobs") return cmd
}
return cmd
} func NewCheckPointJobsCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewJobController(configManager, "checkpoint")
func NewCheckPointJobsCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewJobController(configManager, "checkpoint") cmd := &cobra.Command{
Use: "checkpoint",
cmd := &cobra.Command{ Short: "Checkpoint jobs",
Use: "checkpoint", Long: `Checkpoint jobs by job IDs or filters.
Short: "Checkpoint jobs", Examples:
Long: `Checkpoint jobs by job IDs or filters. xce jobs checkpoint jobid1,jobid2,... # checkpoint jobs by job IDs`,
Examples: RunE: func(cmd *cobra.Command, args []string) error {
xce jobs checkpoint jobid1,jobid2,... # checkpoint jobs by job IDs`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunJobsControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunJobsControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
return cmd
// add control options }
return cmd
} func NewRestartJobsCmd(configManager *config.ConfigManager) *cobra.Command {
restarter := NewJobRestart(configManager)
func NewRestartJobsCmd(configManager *config.ConfigManager) *cobra.Command {
restarter := NewJobRestart(configManager) cmd := &cobra.Command{
Use: "restart",
cmd := &cobra.Command{ Short: "Restart jobs",
Use: "restart", Long: `Restart jobs by job ID.
Short: "Restart jobs", Examples:
Long: `Restart jobs by job ID. xce jobs restart jobid # restart job by jobid
Examples: xce jobs restart -m hostname # restart jobs by hostname`,
xce jobs restart jobid # restart job by jobid RunE: func(cmd *cobra.Command, args []string) error {
xce jobs restart -m hostname # restart jobs by hostname`, if err := restarter.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := restarter.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := restarter.RunJobsRestart(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := restarter.RunJobsRestart(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.StringVarP(&restarter.askedHosts, "hostname", "m", "", "restart on the specific host name")
flags := cmd.Flags() return cmd
flags.StringVarP(&restarter.askedHosts, "hostname", "m", "", "restart on the specific host name") }
return cmd
}

737
src/frontend/xcecli/cmd/jobs/list.go
View file

@ -1,369 +1,368 @@
package jobs package jobs
import ( import (
"fmt" "fmt"
"strings" "strings"
"xcecli/common/output" "xcecli/common/output"
"xcecli/common/request" "xcecli/common/request"
"xcecli/config" "xcecli/config"
job_filter "xcecli/filters" job_filter "xcecli/filters"
"xcecli/models" "xcecli/models"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
var ( var (
resp models.JobsResponse resp models.JobsResponse
loopCount int loopCount int
pageToken string pageToken string
arrayJobsStatsResp models.ArrayJobsStatsResponse arrayJobsStatsResp models.ArrayJobsStatsResponse
) )
// JobsQuerier struct for querying jobs // JobsQuerier struct for querying jobs
type JobsQuerier struct { type JobsQuerier struct {
configManager *config.ConfigManager configManager *config.ConfigManager
filter *job_filter.JobsFilter filter *job_filter.JobsFilter
fields string fields string
jsonOutput bool jsonOutput bool
noHeader bool noHeader bool
verbose bool verbose bool
wideFormat bool // wide format wideFormat bool // wide format
arrayFlag bool // show array job statistics arrayFlag bool // show array job statistics
arrayElementsFlag bool // show array job elements (sub-tasks of array jobs) arrayElementsFlag bool // show array job elements (sub-tasks of array jobs)
} }
// NewJobsQuerier creates a new jobs querier // NewJobsQuerier creates a new jobs querier
func NewJobsQuerier(configManager *config.ConfigManager) *JobsQuerier { func NewJobsQuerier(configManager *config.ConfigManager) *JobsQuerier {
return &JobsQuerier{ return &JobsQuerier{
configManager: configManager, configManager: configManager,
filter: &job_filter.JobsFilter{}, filter: &job_filter.JobsFilter{},
} }
} }
// Parses command line arguments, simplified version that only processes arguments not directly bound to filter // 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 { func (jq *JobsQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error {
// Process JobIDs parameter, which needs special handling // Process JobIDs parameter, which needs special handling
// input multiple jobids, it will use , to connect // input multiple jobids, it will use , to connect
if len(args) > 0 { if len(args) > 0 {
jq.filter.JobIDs = strings.Join(args, ",") jq.filter.JobIDs = strings.Join(args, ",")
} }
// Define mutual exclusions for verbose flag // Define mutual exclusions for verbose flag
if jq.verbose { if jq.verbose {
conflicts := make([]string, 0) conflicts := make([]string, 0)
if jq.wideFormat { if jq.wideFormat {
conflicts = append(conflicts, "--wided (-w)") conflicts = append(conflicts, "--wided (-w)")
} }
if jq.fields != "" { if jq.fields != "" {
conflicts = append(conflicts, "--output (-o)") conflicts = append(conflicts, "--output (-o)")
} }
if len(conflicts) > 0 { if len(conflicts) > 0 {
return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s",
strings.Join(conflicts, ", "), cmd.UsageString()) strings.Join(conflicts, ", "), cmd.UsageString())
} }
} }
return nil return nil
} }
// Run executes the jobs query command // Run executes the jobs query command
func (jq *JobsQuerier) RunJobsList() error { func (jq *JobsQuerier) RunJobsList() error {
// Get server information // Get server information
serverInfo, _ := jq.configManager.GetServerInfo() serverInfo, _ := jq.configManager.GetServerInfo()
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// if array job statistics
// if array job statistics if jq.arrayFlag {
if jq.arrayFlag { if jq.verbose {
if jq.verbose { fmt.Println("not support -A and -l parameters together")
fmt.Println("not support -A and -l parameters together") return nil
return nil }
} return jq.runArrayJobStats(serverInfo, apiClient)
return jq.runArrayJobStats(serverInfo, apiClient) }
}
// if array job elements (sub-tasks of array jobs)
// if array job elements (sub-tasks of array jobs) if jq.arrayElementsFlag {
if jq.arrayElementsFlag { return jq.runArrayJobElements(serverInfo, apiClient)
return jq.runArrayJobElements(serverInfo, apiClient) }
}
// the following is the regular job query logic
// the following is the regular job query logic // Build query parameters
// Build query parameters params := make(map[string]string)
params := make(map[string]string) //fmt.Println(params)
//fmt.Println(params)
// Get filter string
// Get filter string filterStr, err := jq.filter.HandleJobFilter("")
filterStr, err := jq.filter.HandleJobFilter("") if err != nil {
if err != nil { errMsg := request.ExtractErrorMessage(err)
errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to get parameter: [%s]", errMsg)
return fmt.Errorf("failed to get parameter: [%s]", errMsg) } else if filterStr != "" {
} else if filterStr != "" { params["filter"] = filterStr
params["filter"] = filterStr }
}
// Field mapping
// Field mapping fieldsMap := models.JobFieldsMap
fieldsMap := models.JobFieldsMap if jq.verbose && jq.fields == "" {
if jq.verbose && jq.fields == "" { for header := range fieldsMap {
for header := range fieldsMap { jq.fields += header + ","
jq.fields += header + "," }
} jq.fields = strings.TrimSuffix(jq.fields, ",")
jq.fields = strings.TrimSuffix(jq.fields, ",") }
}
// Process field selection
// Process field selection if jq.fields != "" {
if jq.fields != "" { jq.fields = request.CleanFields(jq.fields)
jq.fields = request.CleanFields(jq.fields) mappedFields, err := request.CheckAndMapFields(jq.fields, fieldsMap)
mappedFields, err := request.CheckAndMapFields(jq.fields, fieldsMap) if err != nil {
if err != nil { return err
return err }
} params["fields"] = strings.Join(mappedFields, ",")
params["fields"] = strings.Join(mappedFields, ",") }
}
// create page printer
// create page printer pagePrinter := output.NewPagePrinter(output.DataTypeJobs, output.WithFields(jq.fields),
pagePrinter := output.NewPagePrinter(output.DataTypeJobs, output.WithFields(jq.fields), output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader),
output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat))
output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat))
// record if the first page has been processed
// record if the first page has been processed isFirstPage := true
isFirstPage := true
// Loop to get all pages of data
// Loop to get all pages of data loopCount = 0
loopCount = 0 pageToken = ""
pageToken = ""
for {
for { if loopCount > 0 && pageToken == "" {
if loopCount > 0 && pageToken == "" { break
break }
}
// Only add page_token to params if it's not empty
// Only add page_token to params if it's not empty //if pageToken != "" {
//if pageToken != "" { //params["page_token"] = pageToken
//params["page_token"] = pageToken //}
//}
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/jobs", &resp, params)
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/jobs", &resp, params) if err != nil {
if err != nil { return err
return err }
} if len(resp.Data.Rows) == 0 {
if len(resp.Data.Rows) == 0 { if isFirstPage {
if isFirstPage { fmt.Println("no job found")
fmt.Println("no job found") return nil
return nil }
} }
} // process the first page
// process the first page if isFirstPage {
if isFirstPage { // start printing process
// start printing process pagePrinter.StartPrinting()
pagePrinter.StartPrinting() //
// pagePrinter.PrepareForPrinting(&resp)
pagePrinter.PrepareForPrinting(&resp) isFirstPage = false
isFirstPage = false }
}
// print current page
// print current page pagePrinter.PrintPage(&resp)
pagePrinter.PrintPage(&resp)
// update page token and loop count
// update page token and loop count pageToken = resp.Data.Page_token
pageToken = resp.Data.Page_token
loopCount++
loopCount++ }
}
// finish printing
// finish printing pagePrinter.Finish()
pagePrinter.Finish()
return nil
return nil }
}
// runArrayJobStats process array job statistics
// runArrayJobStats process array job statistics func (jq *JobsQuerier) runArrayJobStats(serverInfo *config.APIServerInfo, apiClient *request.APIClient) error {
func (jq *JobsQuerier) runArrayJobStats(serverInfo *config.APIServerInfo, apiClient *request.APIClient) error { // set query parameters, request array job statistics
// set query parameters, request array job statistics params := map[string]string{
params := map[string]string{ "type": "array_stats",
"type": "array_stats", }
} params["type"] = "array_stats"
params["type"] = "array_stats"
filterStr, err := jq.filter.HandleJobFilter("array_stats")
filterStr, err := jq.filter.HandleJobFilter("array_stats") if err != nil {
if err != nil { errMsg := request.ExtractErrorMessage(err)
errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to get parameter: [%s]", errMsg)
return fmt.Errorf("failed to get parameter: [%s]", errMsg) }
}
if filterStr != "" {
if filterStr != "" { params["filter"] = filterStr
params["filter"] = filterStr }
}
// create response object
// create response object
// create array job page printer
// create array job page printer pagePrinter := output.NewPagePrinter(output.DataTypeArrayJobs, output.WithFields(jq.fields),
pagePrinter := output.NewPagePrinter(output.DataTypeArrayJobs, output.WithFields(jq.fields), output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader),
output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat))
output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat))
// record if the first page has been processed
// record if the first page has been processed isFirstPage := true
isFirstPage := true
// Loop to get all pages of data
// Loop to get all pages of data loopCount = 0
loopCount = 0 pageToken = ""
pageToken = ""
for {
for { if loopCount > 0 && pageToken == "" {
if loopCount > 0 && pageToken == "" { break
break }
}
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/jobs", &arrayJobsStatsResp, params)
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/jobs", &arrayJobsStatsResp, params) if err != nil {
if err != nil { return err
return err }
} if len(arrayJobsStatsResp.Data.Rows) == 0 {
if len(arrayJobsStatsResp.Data.Rows) == 0 { if isFirstPage {
if isFirstPage { fmt.Println("no job found")
fmt.Println("no job found") return nil
return nil }
} }
} // process the first page
// process the first page if isFirstPage {
if isFirstPage { // start printing process
// start printing process pagePrinter.StartPrinting()
pagePrinter.StartPrinting() //
// pagePrinter.PrepareForPrinting(&arrayJobsStatsResp)
pagePrinter.PrepareForPrinting(&arrayJobsStatsResp) isFirstPage = false
isFirstPage = false }
}
// print current page
// print current page pagePrinter.PrintPage(&arrayJobsStatsResp)
pagePrinter.PrintPage(&arrayJobsStatsResp)
// update page token and loop count
// update page token and loop count pageToken = arrayJobsStatsResp.Data.Page_token
pageToken = arrayJobsStatsResp.Data.Page_token
loopCount++
loopCount++ }
} // finish printing
// finish printing pagePrinter.Finish()
pagePrinter.Finish()
return nil
return nil }
}
// runArrayJobElements process array job elements
// runArrayJobElements process array job elements func (jq *JobsQuerier) runArrayJobElements(serverInfo *config.APIServerInfo, apiClient *request.APIClient) error {
func (jq *JobsQuerier) runArrayJobElements(serverInfo *config.APIServerInfo, apiClient *request.APIClient) error { // set query parameters, request array job elements
// set query parameters, request array job elements params := map[string]string{
params := map[string]string{ "filter": "array_idx:ne:0", // query array_idx not equal to 0, that is, the sub-tasks of array jobs
"filter": "array_idx:ne:0", // query array_idx not equal to 0, that is, the sub-tasks of array jobs }
}
// create response object
// create response object var jobsResp models.JobsResponse
var jobsResp models.JobsResponse
// request array job elements data
// request array job elements data err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/xce/v1/jobs?limit=1000", &jobsResp, params)
err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, "/xce/v1/jobs?limit=1000", &jobsResp, params) if err != nil {
if err != nil { return err
return err }
}
// create array job page printer
// create array job page printer pagePrinter := output.NewPagePrinter(output.DataTypeArrayElements, output.WithFields(jq.fields),
pagePrinter := output.NewPagePrinter(output.DataTypeArrayElements, output.WithFields(jq.fields), output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader),
output.WithFormatJson(jq.jsonOutput), output.WithNoHeader(jq.noHeader), output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat))
output.WithVerbose(jq.verbose), output.WithWideFormat(jq.wideFormat)) // start printing process
// start printing process pagePrinter.StartPrinting()
pagePrinter.StartPrinting()
// prepare and print data
// prepare and print data pagePrinter.PrepareForPrinting(&jobsResp)
pagePrinter.PrepareForPrinting(&jobsResp) pagePrinter.PrintPage(&jobsResp)
pagePrinter.PrintPage(&jobsResp)
// finish printing
// finish printing pagePrinter.Finish()
pagePrinter.Finish()
return nil
return nil }
}
func NewListJobsCmd(configManager *config.ConfigManager) *cobra.Command {
func NewListJobsCmd(configManager *config.ConfigManager) *cobra.Command { jobsQuerier := NewJobsQuerier(configManager)
jobsQuerier := NewJobsQuerier(configManager)
// Create the list subcommand
// Create the list subcommand listCmd := &cobra.Command{
listCmd := &cobra.Command{ Use: "list",
Use: "list", Short: "Query jobs info",
Short: "Query jobs info", Long: `Query jobs detail info.
Long: `Query jobs detail info. Example:
Example: xce jobs list # list all jobs
xce jobs list # list all jobs xce jobs list jobid0,jobid1,...jobidn # list jobs by job id
xce jobs list jobid0,jobid1,...jobidn # list jobs by job id xce jobs list -a # list run,done,pending,suspended jobs
xce jobs list -a # list run,done,pending,suspended jobs xce jobs list -d # list done jobs
xce jobs list -d # list done jobs xce jobs list -d -r # list done and running 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 # 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 -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 -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 # 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 -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 -j jobname # list jobs by job name xce jobs list -Jd jobdescription # list jobs by job description
xce jobs list -Jd jobdescription # list jobs by job description xce jobs list --Lp licenseprojectname # list license project jobs
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 ..." # list jobs by fields xce jobs list -o "jobid submit_user ..." --json # list jobs by fields and output in json format
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 # list pending jobs xce jobs list -p -u username # list pending jobs by user
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 -u all # list pending jobs by all users xce jobs list -P projectname # show project jobs
xce jobs list -P projectname # show project jobs xce jobs list -q queuename # list jobs by queue
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 -q queuename -u username # list jobs by user in queue xce jobs list -r # list running jobs
xce jobs list -r # list running jobs xce jobs list -r -u username # list running jobs by user
xce jobs list -r -u username # list running jobs by user xce jobs list -s # list suspended jobs
xce jobs list -s # list suspended jobs xce jobs list -s -u username # list suspended jobs by user
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 -s -u all # list suspended jobs by all users xce jobs list --sla serviceclassname # list jobs by service class
xce jobs list --sla serviceclassname # list jobs by service class xce jobs list -u all # list jobs by all users
xce jobs list -u all # list jobs by all users xce jobs list -u username # list jobs by user
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 all # list jobs by all user xce jobs list -u username -q queuename # list jobs by user and queue
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 -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 -q -u all # list jobs by all users and all queues xce jobs list -A # list array job statistics
xce jobs list -A # list array job statistics xce jobs list -W # list array job elements
xce jobs list -W # list array job elements xce jobs list -m host1 # list jobs executing on specific host
xce jobs list -m host1 # list jobs executing on specific host xce jobs list -m host1,host2 # list jobs executing on multiple hosts`,
xce jobs list -m host1,host2 # list jobs executing on multiple hosts`,
RunE: func(cmd *cobra.Command, args []string) error {
RunE: func(cmd *cobra.Command, args []string) error { if err := jobsQuerier.ParseCmdArgs(cmd, args); err != nil {
if err := jobsQuerier.ParseCmdArgs(cmd, args); err != nil { fmt.Println(strings.ToLower(err.Error()))
fmt.Println(strings.ToLower(err.Error())) return nil
return nil }
} if err := jobsQuerier.RunJobsList(); err != nil {
if err := jobsQuerier.RunJobsList(); err != nil { fmt.Println(strings.ToLower(err.Error()))
fmt.Println(strings.ToLower(err.Error())) return nil
return nil }
} return nil
return nil },
}, }
} // Add command line parameters
// Add command line parameters flags := listCmd.Flags()
flags := listCmd.Flags() flags.BoolVarP(&jobsQuerier.filter.AllJobInfoFlag, "job", "a", false, "show all jobs status")
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.User, "user", "u", "", "filter by user") flags.StringVarP(&jobsQuerier.filter.Queue, "queue", "q", "", "filter by queue")
flags.StringVarP(&jobsQuerier.filter.Queue, "queue", "q", "", "filter by queue") flags.BoolVarP(&jobsQuerier.filter.PendingFlag, "pending", "p", false, "show pending jobs")
flags.BoolVarP(&jobsQuerier.filter.PendingFlag, "pending", "p", false, "show pending jobs") flags.StringVarP(&jobsQuerier.filter.ProjectName, "project", "P", "", "show project 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.RunningFlag, "running", "r", false, "show running jobs") flags.BoolVarP(&jobsQuerier.filter.DoneFlag, "done", "d", false, "show done jobs")
flags.BoolVarP(&jobsQuerier.filter.DoneFlag, "done", "d", false, "show done jobs") flags.BoolVarP(&jobsQuerier.filter.SSUSPFlag, "ssuspFlag", "s", false, "show suspended 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.JobName, "job_name", "J", "", "filter by job name") flags.StringVarP(&jobsQuerier.filter.LicenseProject, "Lp", "", "", "show license project jobs")
flags.StringVarP(&jobsQuerier.filter.LicenseProject, "Lp", "", "", "show license project jobs") flags.StringVarP(&jobsQuerier.filter.JobDescription, "Jd", "", "", "show job description")
flags.StringVarP(&jobsQuerier.filter.JobDescription, "Jd", "", "", "show job description") flags.StringVarP(&jobsQuerier.filter.ServiceClass, "sla", "", "", "show job by service class")
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.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.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.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.StringVarP(&jobsQuerier.fields, "output", "o", "", "output fields for jobs") flags.BoolVarP(&jobsQuerier.wideFormat, "wided", "w", false, "output in wide format")
flags.BoolVarP(&jobsQuerier.wideFormat, "wided", "w", false, "output in wide format") flags.BoolVarP(&jobsQuerier.verbose, "verbose", "l", false, "show job detail")
flags.BoolVarP(&jobsQuerier.verbose, "verbose", "l", false, "show job detail") flags.BoolVarP(&jobsQuerier.jsonOutput, "json", "", false, "output in JSON format")
flags.BoolVarP(&jobsQuerier.jsonOutput, "json", "", false, "output in JSON format") flags.BoolVarP(&jobsQuerier.noHeader, "noheader", "n", false, "output no header")
flags.BoolVarP(&jobsQuerier.noHeader, "noheader", "n", false, "output no header") return listCmd
return listCmd }
}

10
src/frontend/xcecli/cmd/jobs/restart.go
View file

@ -44,23 +44,22 @@ func (c *JobRestart) RunJobsRestart() error {
// Create API client // Create API client
apiClient := request.NewAPIClient(serverInfo.URL) 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) url := fmt.Sprintf("/xce/v1/jobs/%s/restart", c.jobID)
req := &models.JobRestartRequest{ req := &models.JobRestartRequest{
AskedHosts: c.askedHosts, 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) reqData, err := json.Marshal(req)
if err != nil { if err != nil {
errMsg := request.ExtractErrorMessage(err) errMsg := request.ExtractErrorMessage(err)
@ -69,7 +68,6 @@ func (c *JobRestart) printJobRestartResponse(req interface{}, apiClient *request
var resp models.JobRestartResponse var resp models.JobRestartResponse
err = apiClient.PostRequestToAPIServer( err = apiClient.PostRequestToAPIServer(
token, token,
certPath,
url, url,
reqData, reqData,
&resp, &resp,

3
src/frontend/xcecli/cmd/jobs/submit.go
View file

@ -462,7 +462,6 @@ func (js *JobSubmitter) RunJobsSubmit() error {
// create the API client and submit the job // create the API client and submit the job
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
var transferFile []string var transferFile []string
if len(js.uploadFiles) > 0 { if len(js.uploadFiles) > 0 {
@ -517,7 +516,6 @@ func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token strin
err := apiClient.GetRequestFromAPIServer( err := apiClient.GetRequestFromAPIServer(
"Bearer "+token, "Bearer "+token,
"", "",
"",
fmt.Sprintf("/xce/v1/jobs/%s", jobID), fmt.Sprintf("/xce/v1/jobs/%s", jobID),
&resp, &resp,
map[string]string{ /*"fields": "interactive_job_url"*/ }, map[string]string{ /*"fields": "interactive_job_url"*/ },
@ -548,7 +546,6 @@ func (js *JobSubmitter) checkExtendData(apiClient *request.APIClient, token stri
err := apiClient.GetRequestFromAPIServer( err := apiClient.GetRequestFromAPIServer(
"Bearer "+token, "Bearer "+token,
"", "",
"",
fmt.Sprintf("/xce/v1/jobs/%s", jobID), fmt.Sprintf("/xce/v1/jobs/%s", jobID),
&resp, &resp,
map[string]string{"fields": "interactive_job_url"}, map[string]string{"fields": "interactive_job_url"},

3
src/frontend/xcecli/cmd/permissions/list.go
View file

@ -76,14 +76,13 @@ func (pq *PermsQuerier) RunPermsList() error {
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Build query parameters // Build query parameters
params := make(map[string]string) params := make(map[string]string)
pq.fields = []string{} pq.fields = []string{}
// Make API request // 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 { if err != nil {
return err return err
} }

476
src/frontend/xcecli/cmd/queues/controller.go
View file

@ -1,239 +1,237 @@
package queues package queues
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"strings" "strings"
"xcecli/common/request" "xcecli/common/request"
"xcecli/config" "xcecli/config"
"xcecli/models" "xcecli/models"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
// QueueController queue controller // QueueController queue controller
type QueueController struct { type QueueController struct {
configManager *config.ConfigManager configManager *config.ConfigManager
action string action string
queueNames []string queueNames []string
comment string comment string
allQueues bool allQueues bool
} }
// NewQueueController create queue controller // NewQueueController create queue controller
func NewQueueController(configManager *config.ConfigManager, action string) *QueueController { func NewQueueController(configManager *config.ConfigManager, action string) *QueueController {
return &QueueController{ return &QueueController{
configManager: configManager, configManager: configManager,
action: action, action: action,
queueNames: []string{}, queueNames: []string{},
} }
} }
// Run execute queue control // Run execute queue control
func (c *QueueController) RunQueuesControl() error { func (c *QueueController) RunQueuesControl() error {
// get server info and verify authentication // get server info and verify authentication
serverInfo, _ := c.configManager.GetServerInfo() serverInfo, _ := c.configManager.GetServerInfo()
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// handle all queues case
// handle all queues case if c.allQueues {
if c.allQueues { c.queueNames = []string{"all"}
c.queueNames = []string{"all"} }
}
// if no queue names are specified, return an error
// if no queue names are specified, return an error if len(c.queueNames) == 0 {
if len(c.queueNames) == 0 { return fmt.Errorf("must specify queue names or use --all option")
return fmt.Errorf("must specify queue names or use --all option") }
}
// create request
// create request req := models.QueueControlRequest{
req := models.QueueControlRequest{ ControlAction: c.action,
ControlAction: c.action, QueueName: c.queueNames,
QueueName: c.queueNames, Comment: c.comment,
Comment: c.comment, }
}
// convert request to JSON bytes
// convert request to JSON bytes reqData, err := json.Marshal(req)
reqData, err := json.Marshal(req) if err != nil {
if err != nil { errMsg := request.ExtractErrorMessage(err)
errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to encode request data: [%s]", errMsg)
return fmt.Errorf("failed to encode request data: [%s]", errMsg) }
}
// send request
// send request var resp models.QueueControlResponse
var resp models.QueueControlResponse err = apiClient.PostRequestToAPIServer(
err = apiClient.PostRequestToAPIServer( serverInfo.Token,
serverInfo.Token, "/xce/v1/queues/control",
serverInfo.CACert, reqData,
"/xce/v1/queues/control", &resp,
reqData, )
&resp,
) if err != nil {
return err
if err != nil { }
return err
} if len(c.queueNames) == 1 && c.queueNames[0] == "all" {
for _, data := range resp.Data.Accepted {
if len(c.queueNames) == 1 && c.queueNames[0] == "all" { fmt.Printf("queue <%s> is being <%s>\n", data.QueueName, c.action)
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)
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 {
} else { fmt.Printf("queue <%s> is being <%s>\n", data.QueueName, c.action)
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)
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
} }
return nil
} // Parse parse command line arguments
func (c *QueueController) ParseCmdArgs(cmd *cobra.Command, args []string) error {
// Parse parse command line arguments // handle queue name parameters
func (c *QueueController) ParseCmdArgs(cmd *cobra.Command, args []string) error { if len(args) > 0 {
// handle queue name parameters // Split comma-separated queue names
if len(args) > 0 { for _, arg := range args {
// Split comma-separated queue names queues := strings.Split(arg, ",")
for _, arg := range args { c.queueNames = append(c.queueNames, queues...)
queues := strings.Split(arg, ",") }
c.queueNames = append(c.queueNames, queues...) }
}
} return nil
}
return nil
} // NewOpenQueuesCmd create open queues command
func NewOpenQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
// NewOpenQueuesCmd create open queues command controller := NewQueueController(configManager, "open")
func NewOpenQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewQueueController(configManager, "open") cmd := &cobra.Command{
Use: "open",
cmd := &cobra.Command{ Short: "Open queues",
Use: "open", Long: `Open queues by queue name.
Short: "Open queues", Opens queues to accept new job submissions.
Long: `Open queues by queue name.
Opens queues to accept new job submissions. Examples:
xce queues open normal priority # open specific queues
Examples: xce queues open --all # open all queues`,
xce queues open normal priority # open specific queues RunE: func(cmd *cobra.Command, args []string) error {
xce queues open --all # open all queues`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunQueuesControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunQueuesControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues")
flags := cmd.Flags() flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation")
flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues")
flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") return cmd
}
return cmd
} // NewCloseQueuesCmd create close queues command
func NewCloseQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
// NewCloseQueuesCmd create close queues command controller := NewQueueController(configManager, "close")
func NewCloseQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewQueueController(configManager, "close") cmd := &cobra.Command{
Use: "close",
cmd := &cobra.Command{ Short: "Close queues",
Use: "close", Long: `Close queues by queue name.
Short: "Close queues", Closes queues to prevent new job submissions.
Long: `Close queues by queue name.
Closes queues to prevent new job submissions. Examples:
xce queues close normal priority # close specific queues
Examples: xce queues close --all # close all queues`,
xce queues close normal priority # close specific queues RunE: func(cmd *cobra.Command, args []string) error {
xce queues close --all # close all queues`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunQueuesControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunQueuesControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues")
flags := cmd.Flags() flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation")
flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues")
flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") return cmd
}
return cmd
} // NewActiveQueuesCmd create active queues command
func NewActiveQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
// NewActiveQueuesCmd create active queues command controller := NewQueueController(configManager, "active")
func NewActiveQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewQueueController(configManager, "active") cmd := &cobra.Command{
Use: "active",
cmd := &cobra.Command{ Short: "Active queues",
Use: "active", Long: `Active queues by queue name.
Short: "Active queues",
Long: `Active queues by queue name. Examples:
xce queues active # queues_name status close/d`,
Examples: RunE: func(cmd *cobra.Command, args []string) error {
xce queues active # queues_name status close/d`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunQueuesControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunQueuesControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues")
flags := cmd.Flags() flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation")
flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues")
flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") return cmd
}
return cmd
} func NewDeactiveQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewQueueController(configManager, "deactive")
func NewDeactiveQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
controller := NewQueueController(configManager, "deactive") cmd := &cobra.Command{
Use: "deactive",
cmd := &cobra.Command{ Short: "Deactive queues",
Use: "deactive", Long: `Deactive queues by queue name.
Short: "Deactive queues",
Long: `Deactive queues by queue name. Examples:
xce queues deactive # queues_name status close/d`,
Examples: RunE: func(cmd *cobra.Command, args []string) error {
xce queues deactive # queues_name status close/d`, if err := controller.ParseCmdArgs(cmd, args); err != nil {
RunE: func(cmd *cobra.Command, args []string) error { fmt.Println(strings.ToLower(err.Error()))
if err := controller.ParseCmdArgs(cmd, args); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil if err := controller.RunQueuesControl(); err != nil {
} fmt.Println(strings.ToLower(err.Error()))
if err := controller.RunQueuesControl(); err != nil { return nil
fmt.Println(strings.ToLower(err.Error())) }
return nil return nil
} },
return nil }
},
} // add control options
flags := cmd.Flags()
// add control options flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues")
flags := cmd.Flags() flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation")
flags.BoolVarP(&controller.allQueues, "all", "a", false, "apply to all queues")
flags.StringVarP(&controller.comment, "comment", "c", "", "add comment for this operation") return cmd
}
return cmd
}

397
src/frontend/xcecli/cmd/queues/list.go
View file

@ -1,199 +1,198 @@
package queues package queues
import ( import (
"fmt" "fmt"
"strings" "strings"
"xcecli/common/output" "xcecli/common/output"
"xcecli/common/request" "xcecli/common/request"
"xcecli/config" "xcecli/config"
logger "xcecli/config" logger "xcecli/config"
queue_filter "xcecli/filters" queue_filter "xcecli/filters"
"xcecli/models" "xcecli/models"
"github.com/spf13/cobra" "github.com/spf13/cobra"
"go.uber.org/zap" "go.uber.org/zap"
) )
var ( var (
user string user string
queue string queue string
fields string fields string
) )
type QueuesQuerier struct { type QueuesQuerier struct {
configManager *config.ConfigManager configManager *config.ConfigManager
filter *queue_filter.QueuesFilter filter *queue_filter.QueuesFilter
fields string fields string
jsonOutput bool jsonOutput bool
noHeader bool noHeader bool
verbose bool verbose bool
wideFormat bool // wide format wideFormat bool // wide format
} }
func NewQueuesQuerier(configManager *config.ConfigManager) *QueuesQuerier { func NewQueuesQuerier(configManager *config.ConfigManager) *QueuesQuerier {
return &QueuesQuerier{ return &QueuesQuerier{
configManager: configManager, configManager: configManager,
filter: &queue_filter.QueuesFilter{}, filter: &queue_filter.QueuesFilter{},
} }
} }
// Parse parses command line arguments, simplified version that doesn't need to process flags already bound to filter // 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 { func (qq *QueuesQuerier) ParseCmdArgs(cmd *cobra.Command, args []string) error {
// No additional processing needed, all parameters are bound in flags // No additional processing needed, all parameters are bound in flags
if len(args) > 0 { if len(args) > 0 {
qq.filter.QueueName = strings.Join(args, ",") qq.filter.QueueName = strings.Join(args, ",")
} }
// Define mutual exclusions for verbose flag // Define mutual exclusions for verbose flag
if qq.verbose { if qq.verbose {
conflicts := make([]string, 0) conflicts := make([]string, 0)
if qq.wideFormat { if qq.wideFormat {
conflicts = append(conflicts, "--wided (-w)") conflicts = append(conflicts, "--wided (-w)")
} }
if qq.fields != "" { if qq.fields != "" {
conflicts = append(conflicts, "--output (-o)") conflicts = append(conflicts, "--output (-o)")
} }
if len(conflicts) > 0 { if len(conflicts) > 0 {
return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s", return fmt.Errorf("flag --verbose (-l) cannot be used with: %s\n\n%s",
strings.Join(conflicts, ", "), cmd.UsageString()) strings.Join(conflicts, ", "), cmd.UsageString())
} }
} }
return nil return nil
} }
func NewListQueuesCmd(configManager *config.ConfigManager) *cobra.Command { func NewListQueuesCmd(configManager *config.ConfigManager) *cobra.Command {
queuesQuerier := NewQueuesQuerier(configManager) queuesQuerier := NewQueuesQuerier(configManager)
cmd := &cobra.Command{ cmd := &cobra.Command{
Use: "list", Use: "list",
Short: "Query queue info", Short: "Query queue info",
Long: `Query queue info, support filter by queue. Long: `Query queue info, support filter by queue.
example: example:
xce queues list # query all queues xce queues list # query all queues
xce queues list -l # query all queues with detail xce queues list -l # query all queues with detail
xce queues list -w # list queues with wide format`, xce queues list -w # list queues with wide format`,
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
if err := queuesQuerier.ParseCmdArgs(cmd, args); err != nil { if err := queuesQuerier.ParseCmdArgs(cmd, args); err != nil {
fmt.Println(strings.ToLower(err.Error())) fmt.Println(strings.ToLower(err.Error()))
return nil return nil
} }
if err := queuesQuerier.RunQueuesList(); err != nil { if err := queuesQuerier.RunQueuesList(); err != nil {
fmt.Println(strings.ToLower(err.Error())) fmt.Println(strings.ToLower(err.Error()))
return nil return nil
} }
return nil return nil
}, },
} }
// Add command line parameters // Add command line parameters
flags := cmd.Flags() flags := cmd.Flags()
flags.StringVarP(&user, "user", "u", "", "filter by user") flags.StringVarP(&user, "user", "u", "", "filter by user")
flags.StringVarP(&queuesQuerier.filter.QueueName, "queue", "q", "", "filter by queue name") flags.StringVarP(&queuesQuerier.filter.QueueName, "queue", "q", "", "filter by queue name")
flags.StringVarP(&queuesQuerier.fields, "output", "o", "", "output fields for queues") flags.StringVarP(&queuesQuerier.fields, "output", "o", "", "output fields for queues")
flags.BoolVarP(&queuesQuerier.wideFormat, "wided", "w", false, "output in wide format") flags.BoolVarP(&queuesQuerier.wideFormat, "wided", "w", false, "output in wide format")
flags.BoolVarP(&queuesQuerier.verbose, "verbose", "l", false, "show queue detail") flags.BoolVarP(&queuesQuerier.verbose, "verbose", "l", false, "show queue detail")
flags.BoolVarP(&queuesQuerier.jsonOutput, "json", "", false, "output in JSON format") flags.BoolVarP(&queuesQuerier.jsonOutput, "json", "", false, "output in JSON format")
flags.BoolVarP(&queuesQuerier.noHeader, "noheader", "n", false, "output no header") flags.BoolVarP(&queuesQuerier.noHeader, "noheader", "n", false, "output no header")
return cmd return cmd
} }
func (qq *QueuesQuerier) RunQueuesList() error { func (qq *QueuesQuerier) RunQueuesList() error {
// Get configuration // Get configuration
// Get server information // Get server information
serverInfo, _ := qq.configManager.GetServerInfo() serverInfo, _ := qq.configManager.GetServerInfo()
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Create query parameters
// Create query parameters params := make(map[string]string)
params := make(map[string]string)
// Get filter string
// Get filter string filterStr, err := qq.filter.HandleQueueFilter()
filterStr, err := qq.filter.HandleQueueFilter()
if err != nil {
if err != nil { errMsg := request.ExtractErrorMessage(err)
errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to get filter: [%s]", errMsg)
return fmt.Errorf("failed to get filter: [%s]", errMsg) } else if filterStr != "" {
} else if filterStr != "" { params["filter"] = filterStr
params["filter"] = filterStr }
}
// Field mapping
// Field mapping fieldsMap := models.QueueFieldsMap
fieldsMap := models.QueueFieldsMap
if qq.verbose && qq.fields == "" {
if qq.verbose && qq.fields == "" { for header := range fieldsMap {
for header := range fieldsMap { qq.fields += header + ","
qq.fields += header + "," }
} qq.fields = strings.TrimSuffix(qq.fields, ",")
qq.fields = strings.TrimSuffix(qq.fields, ",") }
} // Process field selection
// Process field selection if qq.fields != "" {
if qq.fields != "" { qq.fields = request.CleanFields(qq.fields)
qq.fields = request.CleanFields(qq.fields) mappedFields, err := request.CheckAndMapFields(qq.fields, fieldsMap)
mappedFields, err := request.CheckAndMapFields(qq.fields, fieldsMap) if err != nil {
if err != nil { return err
return err }
} params["fields"] = strings.Join(mappedFields, ",")
params["fields"] = strings.Join(mappedFields, ",") }
}
// create page printer
// create page printer pagePrinter := output.NewPagePrinter(output.DataTypeQueues, output.WithFields(qq.fields),
pagePrinter := output.NewPagePrinter(output.DataTypeQueues, output.WithFields(qq.fields), output.WithFormatJson(qq.jsonOutput), output.WithNoHeader(qq.noHeader),
output.WithFormatJson(qq.jsonOutput), output.WithNoHeader(qq.noHeader), output.WithVerbose(qq.verbose), output.WithWideFormat(qq.wideFormat))
output.WithVerbose(qq.verbose), output.WithWideFormat(qq.wideFormat))
// Record if the first page has been processed
// Record if the first page has been processed isFirstPage := true
isFirstPage := true
var (
var ( resp models.QueuesResponse
resp models.QueuesResponse loopCount int
loopCount int pageToken string
pageToken string )
)
// Loop to get all pages of data
// Loop to get all pages of data for {
for { if loopCount > 0 && pageToken == "" {
if loopCount > 0 && pageToken == "" { break
break }
}
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/queues", &resp, params)
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/queues", &resp, params) if err != nil {
if err != nil { return err
return err }
}
if len(resp.Data.Rows) == 0 {
if len(resp.Data.Rows) == 0 { if isFirstPage {
if isFirstPage { fmt.Println("no queue found")
fmt.Println("no queue found") return nil
return nil }
} //
// }
} // Process the first page
// Process the first page if isFirstPage {
if isFirstPage { // Start printing process
// Start printing process pagePrinter.StartPrinting()
pagePrinter.StartPrinting() // Prepare for printing, calculate column width and table header, etc.
// Prepare for printing, calculate column width and table header, etc. pagePrinter.PrepareForPrinting(&resp)
pagePrinter.PrepareForPrinting(&resp) isFirstPage = false
isFirstPage = false }
} // Print the current page
// Print the current page pagePrinter.PrintPage(&resp)
pagePrinter.PrintPage(&resp)
// Update page token and loop counter
// Update page token and loop counter pageToken = resp.Data.Page_token
pageToken = resp.Data.Page_token loopCount++
loopCount++ }
}
// Finish printing
// Finish printing pagePrinter.Finish()
pagePrinter.Finish()
logger.Debug("queues response completed", zap.Int("pages", loopCount))
logger.Debug("queues response completed", zap.Int("pages", loopCount)) return nil
return nil }
}

6
src/frontend/xcecli/cmd/resources/list.go
View file

@ -135,7 +135,6 @@ func (rq *ResourcesDefQuerier) RunResourcesDefList() error {
serverInfo, _ := rq.configManager.GetServerInfo() serverInfo, _ := rq.configManager.GetServerInfo()
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Create query parameters // Create query parameters
params := make(map[string]string) params := make(map[string]string)
@ -188,7 +187,7 @@ func (rq *ResourcesDefQuerier) RunResourcesDefList() error {
if loopCount > 0 && pageToken == "" { if loopCount > 0 && pageToken == "" {
break 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 { if err != nil {
return err return err
} }
@ -263,7 +262,6 @@ func (rq *ResourcesDistQuerier) RunResourcesDistList() error {
serverInfo, _ := rq.configManager.GetServerInfo() serverInfo, _ := rq.configManager.GetServerInfo()
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Create query parameters // Create query parameters
params := make(map[string]string) params := make(map[string]string)
@ -323,7 +321,7 @@ func (rq *ResourcesDistQuerier) RunResourcesDistList() error {
if loopCount > 0 && pageToken == "" { if loopCount > 0 && pageToken == "" {
break 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 { if err != nil {
return err return err
} }

4
src/frontend/xcecli/cmd/roles/addrole.go
View file

@ -206,7 +206,6 @@ func (rs *RoleSubmitter) RunRolesAdd() error {
serverInfo, _ := rs.configManager.GetServerInfo() serverInfo, _ := rs.configManager.GetServerInfo()
// create the API client and submit the role // create the API client and submit the role
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// build the role request // build the role request
roleReq, err := rs.BuildRoleRequest() roleReq, err := rs.BuildRoleRequest()
@ -223,7 +222,6 @@ func (rs *RoleSubmitter) RunRolesAdd() error {
var resp models.CreateRoleResponse var resp models.CreateRoleResponse
err = apiClient.PostRequestToAPIServer( err = apiClient.PostRequestToAPIServer(
serverInfo.Token, serverInfo.Token,
serverInfo.CACert,
"/xce/v1/roles/create", "/xce/v1/roles/create",
reqData, reqData,
&resp, &resp,
@ -248,7 +246,6 @@ func (rs *RoleSubmitter) RunPermsAdd() error {
serverInfo, _ := rs.configManager.GetServerInfo() serverInfo, _ := rs.configManager.GetServerInfo()
// create the API client and submit the role // create the API client and submit the role
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// build the role request // build the role request
roleReq, err := rs.BuildRoleRequest() roleReq, err := rs.BuildRoleRequest()
@ -267,7 +264,6 @@ func (rs *RoleSubmitter) RunPermsAdd() error {
var resp models.CreateRoleResponse var resp models.CreateRoleResponse
err = apiClient.PostRequestToAPIServer( err = apiClient.PostRequestToAPIServer(
serverInfo.Token, serverInfo.Token,
serverInfo.CACert,
url, url,
reqData, reqData,
&resp, &resp,

2
src/frontend/xcecli/cmd/roles/deleterole.go
View file

@ -65,13 +65,11 @@ func (dr *DeleteRole) RunDeleteRoleInfo() error {
serverInfo, _ := dr.configManager.GetServerInfo() serverInfo, _ := dr.configManager.GetServerInfo()
// create the API client and submit the user // create the API client and submit the user
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
url := fmt.Sprintf("/xce/v1/roles/%s", dr.roleName) url := fmt.Sprintf("/xce/v1/roles/%s", dr.roleName)
var resp request.APIResponse var resp request.APIResponse
err := apiClient.DeleteRequestFromAPIServer( err := apiClient.DeleteRequestFromAPIServer(
serverInfo.Token, serverInfo.Token,
serverInfo.CACert,
url, url,
[]byte(nil), []byte(nil),
&resp, &resp,

3
src/frontend/xcecli/cmd/roles/list.go
View file

@ -88,7 +88,6 @@ func (rq *RolesQuerier) RunRolesInfo(commandType string) error {
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Create query parameters // Create query parameters
params := make(map[string]string) params := make(map[string]string)
@ -156,7 +155,7 @@ func (rq *RolesQuerier) RunRolesInfo(commandType string) error {
break break
} }
resp = models.RolesResponse{} // reset to avoid stale rows from previous page 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 { if err != nil {
return err return err
} }

2
src/frontend/xcecli/cmd/users/assignrole.go
View file

@ -80,7 +80,6 @@ func (us *UserSubmitter) RunUserAssignRole() error {
serverInfo, _ := us.configManager.GetServerInfo() serverInfo, _ := us.configManager.GetServerInfo()
// create the API client and submit the user // create the API client and submit the user
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// build the user request // build the user request
userReq, err := us.BuildUserRequest() userReq, err := us.BuildUserRequest()
@ -98,7 +97,6 @@ func (us *UserSubmitter) RunUserAssignRole() error {
var resp models.UserAssignRoleResponse var resp models.UserAssignRoleResponse
err = apiClient.PostRequestToAPIServer( err = apiClient.PostRequestToAPIServer(
serverInfo.Token, serverInfo.Token,
serverInfo.CACert,
url, url,
reqData, reqData,
&resp, &resp,

6
src/frontend/xcecli/cmd/users/list.go
View file

@ -104,13 +104,12 @@ func (urq *UserRolesQuerier) RunCurrentUserInfo() error {
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Create a response structure for current user info // Create a response structure for current user info
var currentUserResp models.UserResponse // You'll need to define this model var currentUserResp models.UserResponse // You'll need to define this model
// Make API request to get current user info using the token // Make API request to get current user info using the token
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, "/xce/v1/users/current", &currentUserResp, nil) err = apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/xce/v1/users/current", &currentUserResp, nil)
if err != nil { if err != nil {
return fmt.Errorf("failed to get current user info: %w", err) 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 // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Create query parameters // Create query parameters
params := make(map[string]string) params := make(map[string]string)
@ -191,7 +189,7 @@ func (urq *UserRolesQuerier) RunUserRolesInfo(commandType string) error {
break break
} }
resp = models.UserRolesResponse{} // reset to avoid stale rows from previous page 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 { if err != nil {
return err return err
} }

3
src/frontend/xcecli/cmd/users/perm4user.go
View file

@ -84,7 +84,6 @@ func (up *Perms4User) RunUserPermsInfo() error {
// create API client // create API client
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
// Create query parameters // Create query parameters
params := make(map[string]string) params := make(map[string]string)
@ -116,7 +115,7 @@ func (up *Perms4User) RunUserPermsInfo() error {
// Loop to get all pages of data // Loop to get all pages of data
url := "/xce/v1/users/" + up.userName + "/permissions" 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 { if err != nil {
return err return err
} }

2
src/frontend/xcecli/cmd/users/unassignrole.go
View file

@ -73,13 +73,11 @@ func (uu *UnassignUser) RunDeleteUserRoleInfo() error {
serverInfo, _ := uu.configManager.GetServerInfo() serverInfo, _ := uu.configManager.GetServerInfo()
// create the API client and submit the user // create the API client and submit the user
apiClient := request.NewAPIClient(serverInfo.URL) apiClient := request.NewAPIClient(serverInfo.URL)
apiClient.SetRootCAs(serverInfo.CACert)
url := fmt.Sprintf("/xce/v1/users/%s/%s", uu.userName, uu.roleName) url := fmt.Sprintf("/xce/v1/users/%s/%s", uu.userName, uu.roleName)
var resp request.APIResponse var resp request.APIResponse
err := apiClient.DeleteRequestFromAPIServer( err := apiClient.DeleteRequestFromAPIServer(
serverInfo.Token, serverInfo.Token,
serverInfo.CACert,
url, url,
[]byte(nil), []byte(nil),
&resp, &resp,

51
src/frontend/xcecli/common/request/request.go
View file

@ -1,8 +1,6 @@
package request package request
import ( import (
"crypto/tls"
"crypto/x509"
"encoding/json" "encoding/json"
"fmt" "fmt"
@ -45,34 +43,6 @@ func NewAPIClient(baseURL string) *APIClient {
return c 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 // Logon perform the login operation
func (c *APIClient) Logon(username, password string) (*models.LogonResponse, error) { func (c *APIClient) Logon(username, password string) (*models.LogonResponse, error) {
var loginResp models.LogonResponse var loginResp models.LogonResponse
@ -100,13 +70,8 @@ func (c *APIClient) Logon(username, password string) (*models.LogonResponse, err
return &loginResp, nil 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 baseURL := c.BaseURL
if strings.HasPrefix(baseURL, "https://") {
c.Client.SetTLSClientConfig(&tls.Config{
RootCAs: loadRootCAs(certPath),
})
}
if idx := strings.Index(baseURL, dst); idx != -1 { if idx := strings.Index(baseURL, dst); idx != -1 {
baseURL = baseURL[:idx] 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 // 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 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 { if idx := strings.Index(baseURL, dst); idx != -1 {
baseURL = baseURL[:idx] baseURL = baseURL[:idx]
} }
@ -306,13 +266,8 @@ func (c *APIClient) PostRequestToAPIServer(token, certPath, dst string, data []b
return nil 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 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 { if idx := strings.Index(baseURL, dst); idx != -1 {
baseURL = baseURL[:idx] baseURL = baseURL[:idx]
} }

30
src/frontend/xcecli/common/request/request_test.go
View file

@ -2,36 +2,6 @@ package request
import "testing" 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) { func TestAuthHeaderConsistency(t *testing.T) {
token := "test-token-abc" token := "test-token-abc"
headers := map[string]string{ headers := map[string]string{

407
src/frontend/xcecli/config/config.go
View file

@ -1,207 +1,206 @@
package config package config
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"os" "os"
"os/user" "os/user"
"path" "path"
"sort" "sort"
"time" "time"
) )
type APIServerInfo struct { type APIServerInfo struct {
// Name string `json:"name"` // Name string `json:"name"`
URL string `json:"url"` URL string `json:"url"`
Token string `json:"token"` Token string `json:"token"`
Path string `json:"path"` Path string `json:"path"`
CreateTime int64 `json:"createtime"` CreateTime int64 `json:"createtime"`
CACert string `json:"cacert,omitempty"` TokenExpireTime int64 `json:"token_expire_time"`
TokenExpireTime int64 `json:"token_expire_time"` //Version string `json:"version,omitempty"`
//Version string `json:"version,omitempty"` }
}
// APIServerInfoArray is used to sort the APIServerInfo by CreateTime
// APIServerInfoArray is used to sort the APIServerInfo by CreateTime type APIServerInfoArray []APIServerInfo
type APIServerInfoArray []APIServerInfo
func (a APIServerInfoArray) Len() int { return len(a) }
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) 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 }
func (a APIServerInfoArray) Less(i, j int) bool { return a[i].CreateTime > a[j].CreateTime }
type Config struct {
type Config struct { APIServerInfo APIServerInfoArray `json:"servers"`
APIServerInfo APIServerInfoArray `json:"servers"` }
}
// ConfigManager is used to manage the config
// ConfigManager is used to manage the config type ConfigManager struct {
type ConfigManager struct { config *Config
config *Config configPath string
configPath string }
}
// NewConfigManager is used to create the config manager
// NewConfigManager is used to create the config manager func NewConfigManager() (*ConfigManager, error) {
func NewConfigManager() (*ConfigManager, error) { usr, err := user.Current()
usr, err := user.Current() if err != nil {
if err != nil { return nil, fmt.Errorf("failed to get the user info: [%v]", err)
return nil, fmt.Errorf("failed to get the user info: [%v]", err) }
}
configDir := path.Join(usr.HomeDir, ".xce")
configDir := path.Join(usr.HomeDir, ".xce") configPath := path.Join(configDir, "config.json")
configPath := path.Join(configDir, "config.json")
// create the config directory if it does not exist
// create the config directory if it does not exist if err := os.MkdirAll(configDir, 0700); err != nil {
if err := os.MkdirAll(configDir, 0700); err != nil { return nil, fmt.Errorf("failed to create the config directory: [%v]", err)
return nil, fmt.Errorf("failed to create the config directory: [%v]", err) }
}
cm := &ConfigManager{
cm := &ConfigManager{ configPath: configPath,
configPath: configPath, }
}
return cm, nil
return cm, nil }
}
// GetConfig is used to get the config
// GetConfig is used to get the config func (cm *ConfigManager) GetConfig() (*Config, error) {
func (cm *ConfigManager) GetConfig() (*Config, error) { if cm == nil {
if cm == nil { return nil, fmt.Errorf("the config manager is not initialized")
return nil, fmt.Errorf("the config manager is not initialized") }
}
if cm.config != nil {
if cm.config != nil { return cm.config, nil
return cm.config, nil }
}
config, err := cm.loadConfig()
config, err := cm.loadConfig() if err != nil {
if err != nil { return nil, err
return nil, err }
}
cm.config = config
cm.config = config return config, nil
return config, nil }
}
// SaveConfig is used to save the config
// SaveConfig is used to save the config func (cm *ConfigManager) SaveConfig(config *Config) error {
func (cm *ConfigManager) SaveConfig(config *Config) error { // sort the APIServerInfoArray by CreateTime
// sort the APIServerInfoArray by CreateTime sort.Sort(config.APIServerInfo)
sort.Sort(config.APIServerInfo) data, err := json.MarshalIndent(config, "", " ")
data, err := json.MarshalIndent(config, "", " ") if err != nil {
if err != nil { return err
return err }
}
dir := path.Dir(cm.configPath)
dir := path.Dir(cm.configPath) if err := os.MkdirAll(dir, 0700); err != nil {
if err := os.MkdirAll(dir, 0700); err != nil { return err
return err }
}
if err := os.WriteFile(cm.configPath, data, 0600); err != nil {
if err := os.WriteFile(cm.configPath, data, 0600); err != nil { return err
return err }
}
cm.config = config
cm.config = config return nil
return nil }
}
// loadConfig is used to load the config from the file
// loadConfig is used to load the config from the file func (cm *ConfigManager) loadConfig() (*Config, error) {
func (cm *ConfigManager) loadConfig() (*Config, error) { // if the config file does not exist, create the default config
// if the config file does not exist, create the default config if _, err := os.Stat(cm.configPath); os.IsNotExist(err) {
if _, err := os.Stat(cm.configPath); os.IsNotExist(err) { defaultConfig := &Config{
defaultConfig := &Config{ APIServerInfo: make([]APIServerInfo, 0),
APIServerInfo: make([]APIServerInfo, 0), }
} // save the default config
// save the default config if err := cm.SaveConfig(defaultConfig); err != nil {
if err := cm.SaveConfig(defaultConfig); err != nil { return nil, fmt.Errorf("failed to create the default config: [%v]", err)
return nil, fmt.Errorf("failed to create the default config: [%v]", err) }
} return defaultConfig, nil
return defaultConfig, nil }
}
// read the config file
// read the config file data, err := os.ReadFile(cm.configPath)
data, err := os.ReadFile(cm.configPath) if err != nil {
if err != nil { return nil, fmt.Errorf("failed to read the config file: [%v]", err)
return nil, fmt.Errorf("failed to read the config file: [%v]", err) }
}
// parse the config file
// parse the config file var config Config
var config Config if err := json.Unmarshal(data, &config); err != nil {
if err := json.Unmarshal(data, &config); err != nil { return nil, fmt.Errorf("failed to parse the config file: [%s,%v]", cm.configPath, err)
return nil, fmt.Errorf("failed to parse the config file: [%s,%v]", cm.configPath, err) }
}
return &config, nil
return &config, nil }
}
// Save is used to save the config
// Save is used to save the config func (c *Config) Save() error {
func (c *Config) Save() error { usr, err := user.Current()
usr, err := user.Current() if err != nil {
if err != nil { return err
return err }
}
// get the config directory
// get the config directory configDir := path.Join(usr.HomeDir, ".xce")
configDir := path.Join(usr.HomeDir, ".xce") configPath := path.Join(configDir, "config.json")
configPath := path.Join(configDir, "config.json")
// create the config directory if it does not exist
// create the config directory if it does not exist if err := os.MkdirAll(configDir, 0700); err != nil {
if err := os.MkdirAll(configDir, 0700); err != nil { return err
return err }
}
// marshal the config
// marshal the config data, err := json.MarshalIndent(c, "", " ")
data, err := json.MarshalIndent(c, "", " ") if err != nil {
if err != nil { return err
return err }
}
// write the config to the file
// write the config to the file return os.WriteFile(configPath, data, 0600)
return os.WriteFile(configPath, data, 0600) }
}
// IsAuthenticated check if the user has been authenticated by the API server
// IsAuthenticated check if the user has been authenticated by the API server // if not, return an error
// if not, return an error func (cm *ConfigManager) IsAuthenticated() (*APIServerInfo, error) {
func (cm *ConfigManager) IsAuthenticated() (*APIServerInfo, error) { cfg, err := cm.GetConfig()
cfg, err := cm.GetConfig() if err != nil {
if err != nil { return nil, fmt.Errorf("failed to get config: [%v]", err)
return nil, fmt.Errorf("failed to get config: [%v]", err) }
}
if len(cfg.APIServerInfo) == 0 {
if len(cfg.APIServerInfo) == 0 { return nil, fmt.Errorf("you are not logged in, please log in first")
return nil, fmt.Errorf("you are not logged in, please log in first") }
}
serverInfo := &cfg.APIServerInfo[0]
serverInfo := &cfg.APIServerInfo[0] if serverInfo.URL == "" {
if serverInfo.URL == "" { return nil, fmt.Errorf("no default API Server is set, please set a default server")
return nil, fmt.Errorf("no default API Server is set, please set a default server") }
}
if serverInfo.Token == "" {
if serverInfo.Token == "" { return nil, fmt.Errorf("you are not logged in, please log in first")
return nil, fmt.Errorf("you are not logged in, please log in first") }
}
// Use the token expire time from server configuration
// Use the token expire time from server configuration expireTime := serverInfo.TokenExpireTime
expireTime := serverInfo.TokenExpireTime if expireTime == 0 {
if expireTime == 0 { expireTime = 86400 // Default to 24 hours if not set
expireTime = 86400 // Default to 24 hours if not set }
} if time.Now().Unix()-serverInfo.CreateTime > expireTime {
if time.Now().Unix()-serverInfo.CreateTime > expireTime { return nil, fmt.Errorf("your login has expired, please log in again")
return nil, fmt.Errorf("your login has expired, please log in again") }
}
return serverInfo, nil
return serverInfo, nil }
}
// GetServerInfo
// GetServerInfo func (cm *ConfigManager) GetServerInfo() (*APIServerInfo, error) {
func (cm *ConfigManager) GetServerInfo() (*APIServerInfo, error) { cfg, err := cm.GetConfig()
cfg, err := cm.GetConfig() if err != nil {
if err != nil { return nil, fmt.Errorf("failed to get config: [%v]", err)
return nil, fmt.Errorf("failed to get config: [%v]", err) }
}
if len(cfg.APIServerInfo) == 0 || cfg.APIServerInfo[0].URL == "" {
if len(cfg.APIServerInfo) == 0 || cfg.APIServerInfo[0].URL == "" { return nil, fmt.Errorf("not found the API Server information")
return nil, fmt.Errorf("not found the API Server information") }
}
return &cfg.APIServerInfo[0], nil
return &cfg.APIServerInfo[0], nil }
}
// NewConfigManagerWithPath creates a ConfigManager using a specific config file path. // NewConfigManagerWithPath creates a ConfigManager using a specific config file path.
// Used in tests. // Used in tests.

37
src/frontend/xcecli/models/logon.go
View file

@ -1,19 +1,18 @@
package models package models
// LogonData define the data structure in the logon response // LogonData define the data structure in the logon response
type LogonData struct { type LogonData struct {
Token string `json:"token"` Token string `json:"token"`
Path string `json:"path"` Path string `json:"path"`
IP string `json:"ip"` IP string `json:"ip"`
CreateTime int64 `json:"createtime"` CreateTime int64 `json:"createtime"`
TokenExpireTime int64 `json:"token_expire_time"` TokenExpireTime int64 `json:"token_expire_time"`
} }
// LogonResponse define the data structure in the logon response // LogonResponse define the data structure in the logon response
type LogonResponse struct { type LogonResponse struct {
Code int `json:"code"` Code int `json:"code"`
Msg string `json:"msg"` Msg string `json:"msg"`
Data LogonData `json:"data"` Data LogonData `json:"data"`
Count int `json:"count"` Count int `json:"count"`
CACert string `json:"cacert"` }
}

14
src/include/xce/cluster.h
View file

@ -23,8 +23,9 @@ typedef struct {
/** /**
* Cluster connection configuration. * Cluster connection configuration.
* Populated by each caller (worker/scheduler/apiserver) from --peers and * Populated by each caller (worker/scheduler/apiserver) from the --peers
* --ca-cert CLI flags. libxce never reads config files. * 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]. * candidates_count must be in [1, MAX_HA_CANDIDATES].
* A WARN is logged at startup if candidates_count > 3. * A WARN is logged at startup if candidates_count > 3.
@ -32,9 +33,8 @@ typedef struct {
typedef struct { typedef struct {
xce_candidate_t candidates[MAX_HA_CANDIDATES]; xce_candidate_t candidates[MAX_HA_CANDIDATES];
int candidates_count; int candidates_count;
char ca_cert_path[4096]; /* path to CA cert for TLS verify */ char identity[256]; /* "from" name sent to controller */
char identity[256]; /* "from" name sent to controller */ char key[256]; /* authentication key */
char key[256]; /* authentication key */
} xce_cluster_config_t; } xce_cluster_config_t;
/** /**
@ -75,7 +75,6 @@ typedef struct {
* Returns 0 on success, -EINVAL if: * Returns 0 on success, -EINVAL if:
* - candidates_count < 1 or > MAX_HA_CANDIDATES * - candidates_count < 1 or > MAX_HA_CANDIDATES
* - any candidate has empty hostname or port <= 0 * - any candidate has empty hostname or port <= 0
* - ca_cert_path is empty
* - identity is empty * - identity is empty
*/ */
int xce_cluster_config_validate(const xce_cluster_config_t* cfg); 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); 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) */ /* 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, typedef void* (*xce_connect_fn_t)(const char* host, int port);
const char* ca_cert);
void xce_cluster_set_connect_fn(xce_connect_fn_t fn); void xce_cluster_set_connect_fn(xce_connect_fn_t fn);
#endif /* XCE_CLUSTER_H */ #endif /* XCE_CLUSTER_H */

10
src/lib/cluster.c
View file

@ -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 /* 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). */ * 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) 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 */ /* Production path: real SSL connect via netio — stub for now */
(void)host; (void)host;
(void)port; (void)port;
(void)ca_cert;
return NULL; return NULL;
} }
@ -95,8 +94,6 @@ int xce_cluster_config_validate(const xce_cluster_config_t* cfg)
if (cfg->candidates[i].port <= 0) if (cfg->candidates[i].port <= 0)
return -EINVAL; return -EINVAL;
} }
if (cfg->ca_cert_path[0] == '\0')
return -EINVAL;
if (cfg->identity[0] == '\0') if (cfg->identity[0] == '\0')
return -EINVAL; return -EINVAL;
if (cfg->candidates_count > 3) 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++) { for (int i = 0; i < cfg->candidates_count; i++) {
void* ep = do_connect(cfg->candidates[i].hostname, void* ep = do_connect(cfg->candidates[i].hostname,
cfg->candidates[i].port, cfg->candidates[i].port);
cfg->ca_cert_path);
if (ep) if (ep)
return ep; return ep;
} }

5
src/lib/testsuite/test_cluster.c
View file

@ -11,9 +11,8 @@ static int s_connect_fail_n = 0; /* fail first N attempts, then succeed */
static char s_last_host[256] = ""; static char s_last_host[256] = "";
static int s_last_port = 0; 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); strncpy(s_last_host, host, sizeof(s_last_host) - 1);
s_last_port = port; s_last_port = port;
s_connect_call_count++; s_connect_call_count++;
@ -32,8 +31,6 @@ static xce_cluster_config_t make_valid_config(void)
XCE_MAXHOSTNAME_LEN - 1); XCE_MAXHOSTNAME_LEN - 1);
cfg.candidates[1].port = 7001; cfg.candidates[1].port = 7001;
cfg.candidates_count = 2; 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); strncpy(cfg.identity, "worker-01", sizeof(cfg.identity) - 1);
return cfg; return cfg;
} }

3
src/scheduler/include/scheduler.h
View file

@ -122,8 +122,7 @@ typedef struct {
} xce_scheduler_t; } xce_scheduler_t;
extern int sched_init(xce_scheduler_t*, xce_client_config_t*, extern int sched_init(xce_scheduler_t*, xce_client_config_t*,
xce_request_handler_t); const xce_cluster_config_t*, xce_request_handler_t);
extern void sched_set_cluster_cfg(xce_scheduler_t*, const xce_cluster_config_t*);
extern void sched_fini(xce_scheduler_t*); extern void sched_fini(xce_scheduler_t*);
extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*); extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*);
extern int sched_update(xce_scheduler_t*); extern int sched_update(xce_scheduler_t*);

34
src/scheduler/src/main.c
View file

@ -27,15 +27,13 @@ const char* argp_program_bug_address = BUG_EMAIL;
struct arguments { struct arguments {
char* logfile; char* logfile;
int verbose; int verbose;
char* peers; /* --peers value (optional) */ char* peers; /* --peers value (optional) */
char* ca_cert; /* --ca-cert value (optional) */
}; };
static struct argp_option options[] = { static struct argp_option options[] = {
{ "logfile", 'l', "FILE", 0, "The full path of log file" }, { "logfile", 'l', "FILE", 0, "The full path of log file" },
{ "verbose", 'v', 0, 0, "Produce verbose log" }, { "verbose", 'v', 0, 0, "Produce verbose log" },
{ "peers", 'P', "LIST", 0, "HA candidate list: [name=]host:port,..." }, { "peers", 'P', "LIST", 0, "HA candidate list: [name=]host:port,..." },
{ "ca-cert", 'C', "FILE", 0, "CA certificate path for TLS verification" },
{ 0 } { 0 }
}; };
@ -53,9 +51,6 @@ static error_t parse_opt(int key, char* arg, struct argp_state* state)
case 'P': case 'P':
arguments->peers = arg; arguments->peers = arg;
break; break;
case 'C':
arguments->ca_cert = arg;
break;
default: default:
return ARGP_ERR_UNKNOWN; return ARGP_ERR_UNKNOWN;
} }
@ -106,7 +101,9 @@ int main(int argc, char* argv[])
char* loglevel = "info"; char* loglevel = "info";
int logsize = 64 * 1024 * 1024; int logsize = 64 * 1024 * 1024;
int logcount = 10; 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)); 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 */ /* P2.5-6: populate and validate xce_cluster_config_t when --peers given */
if (arguments.peers) { if (arguments.peers) {
xce_cluster_config_t cluster_cfg = {0};
if (xce_parse_peers(arguments.peers, &cluster_cfg) != 0) { if (xce_parse_peers(arguments.peers, &cluster_cfg) != 0) {
fprintf(stderr, fprintf(stderr,
"xce_scheduler: --peers parse error: expected " "xce_scheduler: --peers parse error: expected "
@ -133,16 +128,12 @@ int main(int argc, char* argv[])
arguments.peers); arguments.peers);
return 1; 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", strncpy(cluster_cfg.identity, "xce_scheduler",
sizeof(cluster_cfg.identity) - 1); sizeof(cluster_cfg.identity) - 1);
if (xce_cluster_config_validate(&cluster_cfg) != 0) { if (xce_cluster_config_validate(&cluster_cfg) != 0) {
fprintf(stderr, fprintf(stderr,
"xce_scheduler: invalid cluster config (missing " "xce_scheduler: invalid cluster config (missing "
"--ca-cert?).\n"); "identity?).\n");
return 1; return 1;
} }
if (cluster_cfg.candidates_count > 3) { if (cluster_cfg.candidates_count > 3) {
@ -150,17 +141,16 @@ int main(int argc, char* argv[])
"only <=3 validated for HA", "only <=3 validated for HA",
cluster_cfg.candidates_count); cluster_cfg.candidates_count);
} }
INFO("HA cluster config: %d candidate(s), identity=%s, ca_cert=%s", INFO("HA cluster config: %d candidate(s), identity=%s",
cluster_cfg.candidates_count, cluster_cfg.identity, cluster_cfg.candidates_count, cluster_cfg.identity);
cluster_cfg.ca_cert_path); cluster_cfg_ptr = &cluster_cfg;
sched_set_cluster_cfg(&sched, &cluster_cfg);
} else { } else {
INFO("No --peers flag; using single-controller path"); INFO("No --peers flag; using single-controller path");
} }
init_config(&config); init_config(&config);
sched_init(&sched, &config, scheduler_req_handler); sched_init(&sched, &config, cluster_cfg_ptr, scheduler_req_handler);
signal(SIGINT, sig_handler); signal(SIGINT, sig_handler);

26
src/scheduler/src/scheduler.c
View file

@ -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, int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
const xce_cluster_config_t* cluster_cfg,
xce_request_handler_t handler) xce_request_handler_t handler)
{ {
int rc = 0; int rc = 0;
@ -412,6 +413,10 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
memset(sched, 0x0, sizeof(*sched)); memset(sched, 0x0, sizeof(*sched));
if (cluster_cfg != NULL) {
sched->cluster_cfg = *cluster_cfg;
}
resreq_init(1); /* enable CSE intern table for scheduler process */ resreq_init(1); /* enable CSE intern table for scheduler process */
sched->nextid = 1; 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 */ /* HA mode: iterate candidates until one accepts as primary */
if (sched->cluster_cfg.candidates_count > 0) { if (sched->cluster_cfg.candidates_count > 0) {
int joined = 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++) { for (int i = 0; i < sched->cluster_cfg.candidates_count; i++) {
const char* host = sched->cluster_cfg.candidates[i].hostname; const char* host = sched->cluster_cfg.candidates[i].hostname;
int port = sched->cluster_cfg.candidates[i].port; 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->nmasters = 1;
config->masters[0] = (char*)host; config->masters[0] = (char*)host;
config->master_ports[0] = port; config->master_ports[0] = port;
@ -518,15 +535,6 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
return rc; 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 sched_fini(xce_scheduler_t* sched)
{ {
void* res; void* res;

32
src/worker/src/main.c
View file

@ -289,8 +289,7 @@ static int check_user(void)
} }
/* Long-option identifiers (beyond ASCII range to avoid collisions) */ /* Long-option identifiers (beyond ASCII range to avoid collisions) */
#define OPT_PEERS 256 #define OPT_PEERS 256
#define OPT_CA_CERT 257
int main(int argc, char* argv[]) int main(int argc, char* argv[])
{ {
@ -301,7 +300,6 @@ int main(int argc, char* argv[])
char* custom_config_path = NULL; char* custom_config_path = NULL;
__attribute__((unused)) char worker_log[1024] = { 0 }; __attribute__((unused)) char worker_log[1024] = { 0 };
const char* peers_str = NULL; const char* peers_str = NULL;
const char* ca_cert_str = NULL;
xce_cluster_config_t cluster_cfg = {0}; xce_cluster_config_t cluster_cfg = {0};
int use_ha = 0; /* 1 when --peers given */ int use_ha = 0; /* 1 when --peers given */
@ -394,10 +392,9 @@ int main(int argc, char* argv[])
} }
static struct option long_opts[] = { static struct option long_opts[] = {
{ "peers", required_argument, NULL, OPT_PEERS }, { "peers", required_argument, NULL, OPT_PEERS },
{ "ca-cert", required_argument, NULL, OPT_CA_CERT }, { "help", no_argument, NULL, 'h' },
{ "help", no_argument, NULL, 'h' }, { NULL, 0, NULL, 0 }
{ NULL, 0, NULL, 0 }
}; };
while ((opt = getopt_long(argc, argv, "c:p:m:n:l:h", long_opts, NULL)) 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: case OPT_PEERS:
peers_str = optarg; peers_str = optarg;
break; break;
case OPT_CA_CERT:
ca_cert_str = optarg;
break;
case 'h': case 'h':
default: default:
#ifdef VIRTUAL_CLUSTER #ifdef VIRTUAL_CLUSTER
printf("Usage: %s -n virtual_name [-p remote_port] [-m " printf("Usage: %s -n virtual_name [-p remote_port] [-m "
"remote_host] [-l log_dir] [--peers list] " "remote_host] [-l log_dir] [--peers list] "
"[--ca-cert path] [-h help]\n", "[-h help]\n",
argv[0]); argv[0]);
#else #else
printf("Usage: %s [-p remote_port] [-m remote_host] [-l log_dir] " 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", "[-h help]\n",
argv[0]); argv[0]);
#endif #endif
@ -469,16 +463,12 @@ int main(int argc, char* argv[])
peers_str); peers_str);
return 1; 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, strncpy(cluster_cfg.identity, config.virutal_name,
sizeof(cluster_cfg.identity) - 1); sizeof(cluster_cfg.identity) - 1);
if (xce_cluster_config_validate(&cluster_cfg) != 0) { if (xce_cluster_config_validate(&cluster_cfg) != 0) {
fprintf(stderr, fprintf(stderr,
"xcecd: invalid cluster config (missing ca-cert or " "xcecd: invalid cluster config (missing identity?). "
"identity?). Use --ca-cert and -n.\n"); "Use -n.\n");
return 1; return 1;
} }
if (cluster_cfg.candidates_count > 3) { if (cluster_cfg.candidates_count > 3) {
@ -488,10 +478,8 @@ int main(int argc, char* argv[])
cluster_cfg.candidates_count); cluster_cfg.candidates_count);
} }
XCECD_LOG(XCE_LOG_INFO, XCECD_LOG(XCE_LOG_INFO,
"HA cluster config: %d candidate(s), identity=%s, " "HA cluster config: %d candidate(s), identity=%s",
"ca_cert=%s", cluster_cfg.candidates_count, cluster_cfg.identity);
cluster_cfg.candidates_count, cluster_cfg.identity,
cluster_cfg.ca_cert_path);
use_ha = 1; use_ha = 1;
} else { } else {
XCECD_LOG(XCE_LOG_INFO, XCECD_LOG(XCE_LOG_INFO,