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
}

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

@ -2,11 +2,9 @@ package cmd
import ( import (
"fmt" "fmt"
"net/url"
"os" "os"
"strings" "strings"
"syscall" "syscall"
"xcecli/cmd/auth/cert"
"xcecli/common/request" "xcecli/common/request"
"xcecli/config" "xcecli/config"
logger "xcecli/config" logger "xcecli/config"
@ -19,9 +17,9 @@ import (
) )
// updateServerConfig // updateServerConfig
func updateServerConfig(cfg *config.Config, url, certPath string, loginResp *models.LogonResponse) error { func updateServerConfig(cfg *config.Config, url string, loginResp *models.LogonResponse) error {
logger.Debug("updateServerConfig", zap.String("url", url), zap.String("certPath", certPath)) logger.Debug("updateServerConfig", zap.String("url", url))
logger.Info("updateServerConfig", zap.String("url", url), zap.String("certPath", certPath)) logger.Info("updateServerConfig", zap.String("url", url))
// check if the server with the same URL already exists // check if the server with the same URL already exists
Found := false Found := false
@ -31,7 +29,6 @@ func updateServerConfig(cfg *config.Config, url, certPath string, loginResp *mod
cfg.APIServerInfo[i].Token = loginResp.Data.Token cfg.APIServerInfo[i].Token = loginResp.Data.Token
cfg.APIServerInfo[i].Path = loginResp.Data.Path //save the path cfg.APIServerInfo[i].Path = loginResp.Data.Path //save the path
cfg.APIServerInfo[i].CreateTime = loginResp.Data.CreateTime cfg.APIServerInfo[i].CreateTime = loginResp.Data.CreateTime
cfg.APIServerInfo[i].CACert = certPath
Found = true Found = true
break break
} }
@ -45,7 +42,6 @@ func updateServerConfig(cfg *config.Config, url, certPath string, loginResp *mod
Token: loginResp.Data.Token, Token: loginResp.Data.Token,
Path: loginResp.Data.Path, Path: loginResp.Data.Path,
CreateTime: loginResp.Data.CreateTime, CreateTime: loginResp.Data.CreateTime,
CACert: certPath,
} }
// add it to the end of the array // add it to the end of the array
@ -130,20 +126,6 @@ func runLogon(opts struct{ username, password, url string }, cm *config.ConfigMa
// create the API client // create the API client
apiClient := request.NewAPIClient(opts.url) apiClient := request.NewAPIClient(opts.url)
certPath := ""
// Only setup certs for HTTPS connections
if u, err := url.Parse(opts.url); err == nil && u.Scheme == "https" {
certPath, err = cert.GetCertPath()
apiClient.SetRootCAs(certPath)
if err != nil {
errMsg := request.ExtractErrorMessage(err)
return fmt.Errorf("failed to get cert path: [%s]", errMsg)
}
if err := cert.GeneratorCert(opts.url, certPath); err != nil {
errMsg := request.ExtractErrorMessage(err)
return fmt.Errorf("failed to generate cert: [%s]", errMsg)
}
}
logger.Debug("apiClient", zap.Any("apiClient", apiClient)) logger.Debug("apiClient", zap.Any("apiClient", apiClient))
logger.Info("apiClient", zap.Any("apiClient", apiClient)) logger.Info("apiClient", zap.Any("apiClient", apiClient))
@ -169,7 +151,7 @@ func runLogon(opts struct{ username, password, url string }, cm *config.ConfigMa
logger.Info("loginResp", zap.Any("loginResp", loginResp)) logger.Info("loginResp", zap.Any("loginResp", loginResp))
// update the server config // update the server config
if err := updateServerConfig(cfg, opts.url, certPath, loginResp); err != nil { if err := updateServerConfig(cfg, opts.url, loginResp); err != nil {
errMsg := request.ExtractErrorMessage(err) errMsg := request.ExtractErrorMessage(err)
return fmt.Errorf("failed to update the server config: [%s]", errMsg) return fmt.Errorf("failed to update the server config: [%s]", errMsg)
} }

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

@ -46,7 +46,6 @@ func runClusters(cm *config.ConfigManager) error {
// 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)
@ -61,7 +60,7 @@ func runClusters(cm *config.ConfigManager) error {
// 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, "", serverInfo.CACert, "/xce/v1/clusters", &resp, params) err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/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)

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

@ -41,7 +41,6 @@ func (c *HostController) RunHostsControl() error {
// 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 {
@ -70,7 +69,6 @@ func (c *HostController) RunHostsControl() error {
var resp models.HostControlResponse var resp models.HostControlResponse
err = apiClient.PostRequestToAPIServer( err = apiClient.PostRequestToAPIServer(
serverInfo.Token, serverInfo.Token,
serverInfo.CACert,
"/xce/v1/hosts/control", "/xce/v1/hosts/control",
reqData, reqData,
&resp, &resp,

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

@ -70,7 +70,6 @@ func (hq *HostsQuerier) RunHostsList() error {
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)
@ -124,7 +123,7 @@ func (hq *HostsQuerier) RunHostsList() error {
break break
} }
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/hosts", &resp, params) err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/hosts", &resp, params)
if err != nil { if err != nil {
return err return err
} }

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

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

@ -35,19 +35,18 @@ func (c *JobController) RunJobsControl() error {
// 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, serverInfo.CACert) return c.controlJobsByIDs(apiClient, serverInfo.Token)
} else { } else {
// otherwise use filter conditions // otherwise use filter conditions
return c.controlJobsByFilter(apiClient, serverInfo.Token, serverInfo.CACert) return c.controlJobsByFilter(apiClient, serverInfo.Token)
} }
} }
// control jobs by job IDs // control jobs by job IDs
func (c *JobController) controlJobsByIDs(apiClient *request.APIClient, token string, certPath string) error { func (c *JobController) controlJobsByIDs(apiClient *request.APIClient, token 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))
@ -62,11 +61,11 @@ func (c *JobController) controlJobsByIDs(apiClient *request.APIClient, token str
Action: c.action, Action: c.action,
} }
// encode request data to JSON bytes // encode request data to JSON bytes
return c.printJobControlResponse(req, apiClient, token, certPath, url) return c.printJobControlResponse(req, apiClient, token, url)
} }
// control jobs by filter conditions // control jobs by filter conditions
func (c *JobController) controlJobsByFilter(apiClient *request.APIClient, token string, certPath string) error { func (c *JobController) controlJobsByFilter(apiClient *request.APIClient, token string) error {
url := "/xce/v1/jobs/control" url := "/xce/v1/jobs/control"
req := models.JobControlRequest{ req := models.JobControlRequest{
@ -74,10 +73,10 @@ func (c *JobController) controlJobsByFilter(apiClient *request.APIClient, token
Filter: c.filter, Filter: c.filter,
} }
// encode request data to JSON bytes // encode request data to JSON bytes
return c.printJobControlResponse(req, apiClient, token, certPath, url) return c.printJobControlResponse(req, apiClient, token, url)
} }
func (c *JobController) printJobControlResponse(req interface{}, apiClient *request.APIClient, token string, certPath string, url string) error { func (c *JobController) printJobControlResponse(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)
@ -86,7 +85,6 @@ func (c *JobController) printJobControlResponse(req interface{}, apiClient *requ
var resp models.JobControlResponse var resp models.JobControlResponse
err = apiClient.PostRequestToAPIServer( err = apiClient.PostRequestToAPIServer(
token, token,
certPath,
url, url,
reqData, reqData,
&resp, &resp,

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

@ -76,7 +76,6 @@ func (jq *JobsQuerier) RunJobsList() error {
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 {
@ -147,7 +146,7 @@ func (jq *JobsQuerier) RunJobsList() error {
//params["page_token"] = pageToken //params["page_token"] = pageToken
//} //}
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/jobs", &resp, params) err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/jobs", &resp, params)
if err != nil { if err != nil {
return err return err
} }
@ -218,7 +217,7 @@ func (jq *JobsQuerier) runArrayJobStats(serverInfo *config.APIServerInfo, apiCli
break break
} }
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/jobs", &arrayJobsStatsResp, params) err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/jobs", &arrayJobsStatsResp, params)
if err != nil { if err != nil {
return err return err
} }
@ -262,7 +261,7 @@ func (jq *JobsQuerier) runArrayJobElements(serverInfo *config.APIServerInfo, api
var jobsResp models.JobsResponse var jobsResp models.JobsResponse
// request array job elements data // request array job elements data
err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", serverInfo.CACert, "/xce/v1/jobs?limit=1000", &jobsResp, params) err := apiClient.GetRequestFromAPIServer(serverInfo.Token, "", "/xce/v1/jobs?limit=1000", &jobsResp, params)
if err != nil { if err != nil {
return err return err
} }

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

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

@ -35,7 +35,6 @@ func (c *QueueController) RunQueuesControl() error {
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 {
@ -65,7 +64,6 @@ func (c *QueueController) RunQueuesControl() error {
var resp models.QueueControlResponse var resp models.QueueControlResponse
err = apiClient.PostRequestToAPIServer( err = apiClient.PostRequestToAPIServer(
serverInfo.Token, serverInfo.Token,
serverInfo.CACert,
"/xce/v1/queues/control", "/xce/v1/queues/control",
reqData, reqData,
&resp, &resp,

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

@ -109,7 +109,6 @@ func (qq *QueuesQuerier) RunQueuesList() error {
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)
@ -163,7 +162,7 @@ func (qq *QueuesQuerier) RunQueuesList() error {
break break
} }
err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, serverInfo.CACert, "/xce/v1/queues", &resp, params) err = apiClient.GetRequestFromAPIServer(serverInfo.Token, pageToken, "/xce/v1/queues", &resp, params)
if err != nil { if err != nil {
return err return err
} }

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{

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

@ -16,7 +16,6 @@ type APIServerInfo struct {
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"`
} }

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

@ -15,5 +15,4 @@ type LogonResponse struct {
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"`
} }

10
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,7 +33,6 @@ 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*);

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

@ -28,14 +28,12 @@ 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;
} }
@ -107,6 +102,8 @@ int main(int argc, char* argv[])
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;

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

@ -290,7 +290,6 @@ 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 */
@ -395,7 +393,6 @@ 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 }
}; };
@ -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,