Fix golangci-lint findings across the coordinator and agent
This commit is contained in:
@@ -16,8 +16,8 @@ import (
|
|||||||
func runSetup(args []string) error {
|
func runSetup(args []string) error {
|
||||||
flags := flag.NewFlagSet("setup", flag.ContinueOnError)
|
flags := flag.NewFlagSet("setup", flag.ContinueOnError)
|
||||||
flags.Usage = func() {
|
flags.Usage = func() {
|
||||||
fmt.Fprintf(flags.Output(), "usage: coordinator setup [options]\n")
|
_, _ = fmt.Fprintf(flags.Output(), "usage: coordinator setup [options]\n")
|
||||||
fmt.Fprintf(flags.Output(), "Provisions the coordinator database, schema, and local .env settings.\n\n")
|
_, _ = fmt.Fprintf(flags.Output(), "Provisions the coordinator database, schema, and local .env settings.\n\n")
|
||||||
flags.PrintDefaults()
|
flags.PrintDefaults()
|
||||||
}
|
}
|
||||||
var (
|
var (
|
||||||
@@ -68,7 +68,7 @@ func runSetup(args []string) error {
|
|||||||
log.Error("setup failed", "err", err)
|
log.Error("setup failed", "err", err)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
fmt.Fprint(os.Stdout, summary)
|
_, _ = fmt.Fprint(os.Stdout, summary)
|
||||||
log.Info("setup complete")
|
log.Info("setup complete")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package agent
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
@@ -69,7 +70,7 @@ func (p *WorkerKeyToken) exchangeLocked() error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
request, err := http.NewRequest(http.MethodPost, p.userserviceURL+"/worker-tokens/exchange", bytes.NewReader(payload))
|
request, err := http.NewRequestWithContext(context.Background(), http.MethodPost, p.userserviceURL+"/worker-tokens/exchange", bytes.NewReader(payload))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -79,7 +80,7 @@ func (p *WorkerKeyToken) exchangeLocked() error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("worker key exchange request failed")
|
return fmt.Errorf("worker key exchange request failed")
|
||||||
}
|
}
|
||||||
defer response.Body.Close()
|
defer func() { _ = response.Body.Close() }()
|
||||||
if response.StatusCode != http.StatusOK {
|
if response.StatusCode != http.StatusOK {
|
||||||
return fmt.Errorf("worker key exchange rejected with status %d", response.StatusCode)
|
return fmt.Errorf("worker key exchange rejected with status %d", response.StatusCode)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -11,8 +11,7 @@ import (
|
|||||||
|
|
||||||
func TestWorkerKeyTokenExchangesAndCaches(t *testing.T) {
|
func TestWorkerKeyTokenExchangesAndCaches(t *testing.T) {
|
||||||
var exchanges atomic.Int64
|
var exchanges atomic.Int64
|
||||||
var server *httptest.Server
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
||||||
if r.URL.Path != "/worker-tokens/exchange" {
|
if r.URL.Path != "/worker-tokens/exchange" {
|
||||||
http.NotFound(w, r)
|
http.NotFound(w, r)
|
||||||
return
|
return
|
||||||
@@ -53,8 +52,7 @@ func TestWorkerKeyTokenExchangesAndCaches(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestWorkerKeyTokenRejectsBadKey(t *testing.T) {
|
func TestWorkerKeyTokenRejectsBadKey(t *testing.T) {
|
||||||
var server *httptest.Server
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
||||||
w.WriteHeader(http.StatusUnauthorized)
|
w.WriteHeader(http.StatusUnauthorized)
|
||||||
}))
|
}))
|
||||||
defer server.Close()
|
defer server.Close()
|
||||||
@@ -76,8 +74,7 @@ func TestNewTokenProviderSelectsStrategy(t *testing.T) {
|
|||||||
|
|
||||||
func TestClientRefreshesTokenOnceOn401(t *testing.T) {
|
func TestClientRefreshesTokenOnceOn401(t *testing.T) {
|
||||||
var attempts atomic.Int64
|
var attempts atomic.Int64
|
||||||
var server *httptest.Server
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
||||||
attempts.Add(1)
|
attempts.Add(1)
|
||||||
if attempts.Load() == 1 {
|
if attempts.Load() == 1 {
|
||||||
w.WriteHeader(http.StatusUnauthorized)
|
w.WriteHeader(http.StatusUnauthorized)
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package agent
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
@@ -215,7 +216,7 @@ func (c *Client) Download(uri, destination string) (string, error) {
|
|||||||
if resolved.Scheme != "http" && resolved.Scheme != "https" {
|
if resolved.Scheme != "http" && resolved.Scheme != "https" {
|
||||||
return "", fmt.Errorf("input URI must be an HTTP(S) URL")
|
return "", fmt.Errorf("input URI must be an HTTP(S) URL")
|
||||||
}
|
}
|
||||||
request, err := http.NewRequest(http.MethodGet, resolved.String(), nil)
|
request, err := http.NewRequestWithContext(context.Background(), http.MethodGet, resolved.String(), nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
@@ -230,17 +231,18 @@ func (c *Client) Download(uri, destination string) (string, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return "", &TransientError{msg: "input download failed"}
|
return "", &TransientError{msg: "input download failed"}
|
||||||
}
|
}
|
||||||
defer response.Body.Close()
|
defer func() { _ = response.Body.Close() }()
|
||||||
if response.StatusCode == http.StatusUnauthorized && c.refreshAndRetry() {
|
if response.StatusCode == http.StatusUnauthorized && c.refreshAndRetry() {
|
||||||
return c.Download(uri, destination)
|
return c.Download(uri, destination)
|
||||||
}
|
}
|
||||||
if response.StatusCode != http.StatusOK {
|
if response.StatusCode != http.StatusOK {
|
||||||
return "", &CoordinatorError{msg: fmt.Sprintf("input download rejected with status %d", response.StatusCode)}
|
return "", &CoordinatorError{msg: fmt.Sprintf("input download rejected with status %d", response.StatusCode)}
|
||||||
}
|
}
|
||||||
if err := os.MkdirAll(filepath.Dir(destination), 0o755); err != nil {
|
if err := os.MkdirAll(filepath.Dir(destination), 0o750); err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
target, err := os.OpenFile(destination, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
|
// #nosec G304 -- destination is the worker's own attempt directory file.
|
||||||
|
target, err := os.OpenFile(destination, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
@@ -248,7 +250,7 @@ func (c *Client) Download(uri, destination string) (string, error) {
|
|||||||
_, copyErr := io.Copy(io.MultiWriter(target, digest), response.Body)
|
_, copyErr := io.Copy(io.MultiWriter(target, digest), response.Body)
|
||||||
closeErr := target.Close()
|
closeErr := target.Close()
|
||||||
if copyErr != nil {
|
if copyErr != nil {
|
||||||
os.Remove(destination)
|
_ = os.Remove(destination)
|
||||||
return "", &TransientError{msg: "input download interrupted"}
|
return "", &TransientError{msg: "input download interrupted"}
|
||||||
}
|
}
|
||||||
if closeErr != nil {
|
if closeErr != nil {
|
||||||
@@ -259,29 +261,30 @@ func (c *Client) Download(uri, destination string) (string, error) {
|
|||||||
|
|
||||||
// Upload streams a partial artifact and verifies the returned metadata.
|
// Upload streams a partial artifact and verifies the returned metadata.
|
||||||
func (c *Client) Upload(task *Task, workerID string, path, contentType string) (*Uploaded, error) {
|
func (c *Client) Upload(task *Task, workerID string, path, contentType string) (*Uploaded, error) {
|
||||||
|
// #nosec G304 -- the upload path is this worker's own artifact file.
|
||||||
file, err := os.Open(path)
|
file, err := os.Open(path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
info, err := file.Stat()
|
info, err := file.Stat()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
file.Close()
|
_ = file.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
digest := sha256.New()
|
digest := sha256.New()
|
||||||
if _, err := io.Copy(digest, file); err != nil {
|
if _, err := io.Copy(digest, file); err != nil {
|
||||||
file.Close()
|
_ = file.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if _, err := file.Seek(0, io.SeekStart); err != nil {
|
if _, err := file.Seek(0, io.SeekStart); err != nil {
|
||||||
file.Close()
|
_ = file.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
localSHA := hex.EncodeToString(digest.Sum(nil))
|
localSHA := hex.EncodeToString(digest.Sum(nil))
|
||||||
uploadURL := c.baseURL + "/tasks/" + url.PathEscape(task.TaskID) + "/artifacts/" + url.PathEscape(filepath.Base(path))
|
uploadURL := c.baseURL + "/tasks/" + url.PathEscape(task.TaskID) + "/artifacts/" + url.PathEscape(filepath.Base(path))
|
||||||
request, err := http.NewRequest(http.MethodPut, uploadURL, file)
|
request, err := http.NewRequestWithContext(context.Background(), http.MethodPut, uploadURL, file)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
file.Close()
|
_ = file.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
request.ContentLength = info.Size()
|
request.ContentLength = info.Size()
|
||||||
@@ -290,18 +293,18 @@ func (c *Client) Upload(task *Task, workerID string, path, contentType string) (
|
|||||||
request.Header.Set("X-Task-Attempt", strconv.Itoa(task.Attempt))
|
request.Header.Set("X-Task-Attempt", strconv.Itoa(task.Attempt))
|
||||||
headers, err := c.authHeaders()
|
headers, err := c.authHeaders()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
file.Close()
|
_ = file.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
for name, value := range headers {
|
for name, value := range headers {
|
||||||
request.Header.Set(name, value)
|
request.Header.Set(name, value)
|
||||||
}
|
}
|
||||||
response, err := c.apiClient.Do(request)
|
response, err := c.apiClient.Do(request)
|
||||||
file.Close()
|
_ = file.Close()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, &TransientError{msg: "artifact upload failed"}
|
return nil, &TransientError{msg: "artifact upload failed"}
|
||||||
}
|
}
|
||||||
defer response.Body.Close()
|
defer func() { _ = response.Body.Close() }()
|
||||||
raw, err := io.ReadAll(io.LimitReader(response.Body, 1<<20))
|
raw, err := io.ReadAll(io.LimitReader(response.Body, 1<<20))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, &TransientError{msg: "artifact upload interrupted"}
|
return nil, &TransientError{msg: "artifact upload interrupted"}
|
||||||
@@ -334,7 +337,7 @@ func (c *Client) requestJSON(method, path string, payload any) (int, map[string]
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, nil, err
|
return 0, nil, err
|
||||||
}
|
}
|
||||||
request, err := http.NewRequest(method, c.baseURL+path, bytes.NewReader(body))
|
request, err := http.NewRequestWithContext(context.Background(), method, c.baseURL+path, bytes.NewReader(body))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, nil, err
|
return 0, nil, err
|
||||||
}
|
}
|
||||||
@@ -350,7 +353,7 @@ func (c *Client) requestJSON(method, path string, payload any) (int, map[string]
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, nil, &TransientError{msg: "coordinator request failed"}
|
return 0, nil, &TransientError{msg: "coordinator request failed"}
|
||||||
}
|
}
|
||||||
defer response.Body.Close()
|
defer func() { _ = response.Body.Close() }()
|
||||||
raw, err := io.ReadAll(io.LimitReader(response.Body, 1<<20))
|
raw, err := io.ReadAll(io.LimitReader(response.Body, 1<<20))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, nil, &TransientError{msg: "coordinator request interrupted"}
|
return 0, nil, &TransientError{msg: "coordinator request interrupted"}
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package agent
|
|||||||
import (
|
import (
|
||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
@@ -20,7 +21,8 @@ func newTestClient(t *testing.T, server *httptest.Server) *Client {
|
|||||||
|
|
||||||
func TestClientRegisterClaimHeartbeat(t *testing.T) {
|
func TestClientRegisterClaimHeartbeat(t *testing.T) {
|
||||||
var registered, claimed, heartbeated bool
|
var registered, claimed, heartbeated bool
|
||||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
var server *httptest.Server //nolint:staticcheck // the handler closure references server
|
||||||
|
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
if r.Header.Get("Authorization") != "Bearer test-token" {
|
if r.Header.Get("Authorization") != "Bearer test-token" {
|
||||||
http.Error(w, "missing token", http.StatusUnauthorized)
|
http.Error(w, "missing token", http.StatusUnauthorized)
|
||||||
return
|
return
|
||||||
@@ -76,7 +78,8 @@ func TestClientRegisterClaimHeartbeat(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestClientClaimEmptyAndConflict(t *testing.T) {
|
func TestClientClaimEmptyAndConflict(t *testing.T) {
|
||||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
var server *httptest.Server //nolint:staticcheck // the handler closure references server
|
||||||
|
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
switch r.URL.Path {
|
switch r.URL.Path {
|
||||||
case "/tasks/claim":
|
case "/tasks/claim":
|
||||||
w.WriteHeader(http.StatusNoContent)
|
w.WriteHeader(http.StatusNoContent)
|
||||||
@@ -103,14 +106,14 @@ func TestClientClaimEmptyAndConflict(t *testing.T) {
|
|||||||
}
|
}
|
||||||
if _, err := client.Heartbeat(claimed, "worker"); err == nil {
|
if _, err := client.Heartbeat(claimed, "worker"); err == nil {
|
||||||
t.Error("expected conflict error")
|
t.Error("expected conflict error")
|
||||||
} else if _, ok := err.(*ConflictError); !ok {
|
} else if !errors.As(err, &conflictError) {
|
||||||
t.Errorf("error type = %T", err)
|
t.Errorf("error type = %T", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestClientUploadSubmitFail(t *testing.T) {
|
func TestClientUploadSubmitFail(t *testing.T) {
|
||||||
var uploadedPath string
|
var uploadedPath string
|
||||||
var server *httptest.Server
|
var server *httptest.Server //nolint:staticcheck // the handler closure references server
|
||||||
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
switch {
|
switch {
|
||||||
case r.Method == http.MethodPut && strings.HasPrefix(r.URL.Path, "/tasks/11111111-1111-4111-8111-111111111111/artifacts/"):
|
case r.Method == http.MethodPut && strings.HasPrefix(r.URL.Path, "/tasks/11111111-1111-4111-8111-111111111111/artifacts/"):
|
||||||
@@ -162,14 +165,12 @@ func TestClientUploadSubmitFail(t *testing.T) {
|
|||||||
|
|
||||||
func TestClientDownloadVerifiesChecksumAndStripsAuthOnRedirect(t *testing.T) {
|
func TestClientDownloadVerifiesChecksumAndStripsAuthOnRedirect(t *testing.T) {
|
||||||
var redirectedAuth string
|
var redirectedAuth string
|
||||||
var bucket *httptest.Server
|
bucket := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
bucket = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
||||||
redirectedAuth = r.Header.Get("Authorization")
|
redirectedAuth = r.Header.Get("Authorization")
|
||||||
_, _ = w.Write([]byte("input bytes"))
|
_, _ = w.Write([]byte("input bytes"))
|
||||||
}))
|
}))
|
||||||
defer bucket.Close()
|
defer bucket.Close()
|
||||||
var server *httptest.Server
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
||||||
if r.URL.Path == "/tasks/11111111-1111-4111-8111-111111111111/input" {
|
if r.URL.Path == "/tasks/11111111-1111-4111-8111-111111111111/input" {
|
||||||
http.Redirect(w, r, bucket.URL+"/presigned", http.StatusFound)
|
http.Redirect(w, r, bucket.URL+"/presigned", http.StatusFound)
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -1,10 +1,8 @@
|
|||||||
package agent
|
package agent
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"crypto/sha256"
|
"errors"
|
||||||
"encoding/hex"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -21,6 +19,9 @@ type Outcome struct {
|
|||||||
|
|
||||||
// Daemon is the agent state machine: register, claim, execute via the Python
|
// Daemon is the agent state machine: register, claim, execute via the Python
|
||||||
// task runner, upload, and submit — mirroring the Python worker's lifecycle.
|
// task runner, upload, and submit — mirroring the Python worker's lifecycle.
|
||||||
|
// conflictError is the errors.As target for lease conflicts.
|
||||||
|
var conflictError *ConflictError
|
||||||
|
|
||||||
type Daemon struct {
|
type Daemon struct {
|
||||||
config *Config
|
config *Config
|
||||||
client *Client
|
client *Client
|
||||||
@@ -159,14 +160,14 @@ func (d *Daemon) runOnce() (Outcome, error) {
|
|||||||
}
|
}
|
||||||
started := time.Now()
|
started := time.Now()
|
||||||
taskDir := filepath.Join(d.config.WorkDir, task.TaskID, fmt.Sprint(task.Attempt))
|
taskDir := filepath.Join(d.config.WorkDir, task.TaskID, fmt.Sprint(task.Attempt))
|
||||||
if err := os.MkdirAll(taskDir, 0o755); err != nil {
|
if err := os.MkdirAll(taskDir, 0o750); err != nil {
|
||||||
return Outcome{Claimed: true}, err
|
return Outcome{Claimed: true}, err
|
||||||
}
|
}
|
||||||
heartbeat := newLeaseHeartbeat(task, workerID, d.client, d.config.Heartbeat)
|
heartbeat := newLeaseHeartbeat(task, workerID, d.client, d.config.Heartbeat)
|
||||||
completed := false
|
completed := false
|
||||||
err = heartbeat.Start()
|
err = heartbeat.Start()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if _, ok := err.(*ConflictError); ok {
|
if errors.As(err, &conflictError) {
|
||||||
d.log.Warn("lease lost", "task_id", task.TaskID)
|
d.log.Warn("lease lost", "task_id", task.TaskID)
|
||||||
return Outcome{Claimed: true}, nil
|
return Outcome{Claimed: true}, nil
|
||||||
}
|
}
|
||||||
@@ -178,7 +179,7 @@ func (d *Daemon) runOnce() (Outcome, error) {
|
|||||||
// attempt directories are retained under the work directory.
|
// attempt directories are retained under the work directory.
|
||||||
failure := d.executeTask(task, workerID, taskDir, started, heartbeat)
|
failure := d.executeTask(task, workerID, taskDir, started, heartbeat)
|
||||||
if failure != nil {
|
if failure != nil {
|
||||||
if _, ok := failure.(*ConflictError); ok {
|
if errors.As(failure, &conflictError) {
|
||||||
d.log.Warn("lease lost", "task_id", task.TaskID)
|
d.log.Warn("lease lost", "task_id", task.TaskID)
|
||||||
return Outcome{Claimed: true}, nil
|
return Outcome{Claimed: true}, nil
|
||||||
}
|
}
|
||||||
@@ -235,17 +236,17 @@ func (d *Daemon) executeTask(task *Task, workerID, taskDir string, started time.
|
|||||||
|
|
||||||
func (d *Daemon) reportFailure(task *Task, workerID string, failure error) {
|
func (d *Daemon) reportFailure(task *Task, workerID string, failure error) {
|
||||||
var code string
|
var code string
|
||||||
switch failure.(type) {
|
var coordinatorErr *CoordinatorError
|
||||||
case *CoordinatorError:
|
if errors.As(failure, &coordinatorErr) {
|
||||||
code = "ValueError"
|
code = "ValueError"
|
||||||
default:
|
} else {
|
||||||
code = "TaskRunnerFailed"
|
code = "TaskRunnerFailed"
|
||||||
}
|
}
|
||||||
retryable := IsRetryableError(failure)
|
retryable := IsRetryableError(failure)
|
||||||
message := SanitizeErrorMessage(failure.Error(), d.config.WorkDir)
|
message := SanitizeErrorMessage(failure.Error(), d.config.WorkDir)
|
||||||
d.log.Warn("task failed", "task_id", task.TaskID, "error_code", code, "retryable", retryable)
|
d.log.Warn("task failed", "task_id", task.TaskID, "error_code", code, "retryable", retryable)
|
||||||
if err := d.client.Fail(task, workerID, code, message, retryable); err != nil {
|
if err := d.client.Fail(task, workerID, code, message, retryable); err != nil {
|
||||||
if _, ok := err.(*ConflictError); ok {
|
if errors.As(err, &conflictError) {
|
||||||
d.log.Warn("lease lost while reporting failure", "task_id", task.TaskID)
|
d.log.Warn("lease lost while reporting failure", "task_id", task.TaskID)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -333,17 +334,3 @@ func (h *leaseHeartbeat) nextDelay() time.Duration {
|
|||||||
func roundSeconds(seconds float64) float64 {
|
func roundSeconds(seconds float64) float64 {
|
||||||
return float64(int64(seconds*1000)) / 1000
|
return float64(int64(seconds*1000)) / 1000
|
||||||
}
|
}
|
||||||
|
|
||||||
// File-digest helper used by tests.
|
|
||||||
func sha256File(path string) (string, error) {
|
|
||||||
file, err := os.Open(path)
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
defer file.Close()
|
|
||||||
digest := sha256.New()
|
|
||||||
if _, err := io.Copy(digest, file); err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
return hex.EncodeToString(digest.Sum(nil)), nil
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package agent
|
package agent
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"regexp"
|
"regexp"
|
||||||
@@ -36,10 +37,9 @@ func IsRetryableError(err error) bool {
|
|||||||
if err == nil {
|
if err == nil {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
switch err.(type) {
|
var coordinatorErr *CoordinatorError
|
||||||
case *CoordinatorError:
|
var pathErr *os.PathError
|
||||||
return false
|
if errors.As(err, &coordinatorErr) || errors.As(err, &pathErr) {
|
||||||
case *os.PathError:
|
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
return true
|
return true
|
||||||
|
|||||||
@@ -2,7 +2,9 @@ package agent
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"os/exec"
|
"os/exec"
|
||||||
@@ -23,7 +25,7 @@ func NewTaskRunner(command []string) *TaskRunner {
|
|||||||
// directory, the Python entry computes and seals the partial, and the written
|
// directory, the Python entry computes and seals the partial, and the written
|
||||||
// manifest is parsed back. stderr is captured for failure reporting.
|
// manifest is parsed back. stderr is captured for failure reporting.
|
||||||
func (r *TaskRunner) Run(task *Task, taskDir string, manifestPath string, extraEnv []string) (*TaskRunnerManifest, error) {
|
func (r *TaskRunner) Run(task *Task, taskDir string, manifestPath string, extraEnv []string) (*TaskRunnerManifest, error) {
|
||||||
if err := os.MkdirAll(taskDir, 0o755); err != nil {
|
if err := os.MkdirAll(taskDir, 0o750); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
payload := map[string]any{
|
payload := map[string]any{
|
||||||
@@ -51,17 +53,20 @@ func (r *TaskRunner) Run(task *Task, taskDir string, manifestPath string, extraE
|
|||||||
"--task-dir", taskDir,
|
"--task-dir", taskDir,
|
||||||
"--output", manifestPath,
|
"--output", manifestPath,
|
||||||
)
|
)
|
||||||
command := exec.Command(r.command[0], args...)
|
// #nosec G204 -- the command comes from the operator-configured TASK_RUNNER.
|
||||||
|
command := exec.CommandContext(context.Background(), r.command[0], args...)
|
||||||
command.Dir = taskDir
|
command.Dir = taskDir
|
||||||
command.Env = append(os.Environ(), extraEnv...)
|
command.Env = append(os.Environ(), extraEnv...)
|
||||||
var stderr bytes.Buffer
|
var stderr bytes.Buffer
|
||||||
command.Stderr = &stderr
|
command.Stderr = &stderr
|
||||||
if err := command.Run(); err != nil {
|
if err := command.Run(); err != nil {
|
||||||
if exitErr, ok := err.(*exec.ExitError); ok {
|
var exitErr *exec.ExitError
|
||||||
|
if errors.As(err, &exitErr) {
|
||||||
return nil, runnerExitError(exitErr.ExitCode(), stderr.String())
|
return nil, runnerExitError(exitErr.ExitCode(), stderr.String())
|
||||||
}
|
}
|
||||||
return nil, fmt.Errorf("task runner could not be started: %w", err)
|
return nil, fmt.Errorf("task runner could not be started: %w", err)
|
||||||
}
|
}
|
||||||
|
// #nosec G304 -- the manifest path is inside the worker's own task directory.
|
||||||
raw, err := os.ReadFile(manifestPath)
|
raw, err := os.ReadFile(manifestPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("task runner produced no result manifest")
|
return nil, fmt.Errorf("task runner produced no result manifest")
|
||||||
|
|||||||
@@ -58,7 +58,7 @@ func Run(ctx context.Context, options Options) (string, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
report := func(format string, args ...any) {
|
report := func(format string, args ...any) {
|
||||||
fmt.Fprintf(options.Out, format+"\n", args...)
|
_, _ = fmt.Fprintf(options.Out, format+"\n", args...)
|
||||||
}
|
}
|
||||||
|
|
||||||
report("SciMesh coordinator setup")
|
report("SciMesh coordinator setup")
|
||||||
@@ -195,6 +195,7 @@ func writeEnvFile(options Options, secret, storageDir string) error {
|
|||||||
"COORDINATOR_STORAGE_DIR=" + storageDir,
|
"COORDINATOR_STORAGE_DIR=" + storageDir,
|
||||||
"", // trailing newline
|
"", // trailing newline
|
||||||
}, "\n")
|
}, "\n")
|
||||||
|
// #nosec G703 -- the env file path is operator-supplied (--env-file / ENV_FILE).
|
||||||
if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
|
if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
|
||||||
return fmt.Errorf("write %s: %w", path, err)
|
return fmt.Errorf("write %s: %w", path, err)
|
||||||
}
|
}
|
||||||
@@ -224,7 +225,7 @@ func SanitizeDatabaseURL(raw string) string {
|
|||||||
|
|
||||||
// prompt asks a question and returns the trimmed answer ("" on EOF).
|
// prompt asks a question and returns the trimmed answer ("" on EOF).
|
||||||
func prompt(options Options, question, fallback string) string {
|
func prompt(options Options, question, fallback string) string {
|
||||||
fmt.Fprintf(options.Out, "%s [%s]: ", question, fallback)
|
_, _ = fmt.Fprintf(options.Out, "%s [%s]: ", question, fallback)
|
||||||
reader := bufio.NewReader(options.In)
|
reader := bufio.NewReader(options.In)
|
||||||
line, err := reader.ReadString('\n')
|
line, err := reader.ReadString('\n')
|
||||||
if err != nil && !errors.Is(err, io.EOF) {
|
if err != nil && !errors.Is(err, io.EOF) {
|
||||||
|
|||||||
@@ -328,6 +328,7 @@ func newJobPayload(catalog *workloads.Catalog) template.JS {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return template.JS("null")
|
return template.JS("null")
|
||||||
}
|
}
|
||||||
|
// #nosec G203 -- the payload is marshaled JSON from the embedded workload catalog, injected as script data.
|
||||||
return template.JS(encoded)
|
return template.JS(encoded)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package http
|
package http
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"io"
|
"io"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -34,13 +35,13 @@ func TestUIDocsServesIndexAndNestedFiles(t *testing.T) {
|
|||||||
server := docsTestServer(t, root)
|
server := docsTestServer(t, root)
|
||||||
|
|
||||||
index := httptest.NewRecorder()
|
index := httptest.NewRecorder()
|
||||||
server.handleUIDocs(index, httptest.NewRequest(http.MethodGet, "/ui/docs/", nil))
|
server.handleUIDocs(index, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/", nil))
|
||||||
if index.Code != http.StatusOK || !strings.Contains(index.Body.String(), "<h1>Home</h1>") {
|
if index.Code != http.StatusOK || !strings.Contains(index.Body.String(), "<h1>Home</h1>") {
|
||||||
t.Fatalf("index = %d %q", index.Code, index.Body.String())
|
t.Fatalf("index = %d %q", index.Code, index.Body.String())
|
||||||
}
|
}
|
||||||
|
|
||||||
page := httptest.NewRecorder()
|
page := httptest.NewRecorder()
|
||||||
server.handleUIDocs(page, httptest.NewRequest(http.MethodGet, "/ui/docs/api/page.html", nil))
|
server.handleUIDocs(page, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/api/page.html", nil))
|
||||||
if page.Code != http.StatusOK || !strings.Contains(page.Body.String(), "<h1>API page</h1>") {
|
if page.Code != http.StatusOK || !strings.Contains(page.Body.String(), "<h1>API page</h1>") {
|
||||||
t.Fatalf("nested page = %d %q", page.Code, page.Body.String())
|
t.Fatalf("nested page = %d %q", page.Code, page.Body.String())
|
||||||
}
|
}
|
||||||
@@ -54,7 +55,7 @@ func TestUIDocsRejectsPathTraversal(t *testing.T) {
|
|||||||
}
|
}
|
||||||
server := docsTestServer(t, root)
|
server := docsTestServer(t, root)
|
||||||
|
|
||||||
request := httptest.NewRequest(http.MethodGet, "/ui/docs/../secret.txt", nil)
|
request := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/../secret.txt", nil)
|
||||||
request.URL.Path = "/ui/docs/../secret.txt"
|
request.URL.Path = "/ui/docs/../secret.txt"
|
||||||
recorder := httptest.NewRecorder()
|
recorder := httptest.NewRecorder()
|
||||||
server.handleUIDocs(recorder, request)
|
server.handleUIDocs(recorder, request)
|
||||||
@@ -66,14 +67,14 @@ func TestUIDocsRejectsPathTraversal(t *testing.T) {
|
|||||||
func TestUIDocsShowsBuildHintWhenDisabledOrMissing(t *testing.T) {
|
func TestUIDocsShowsBuildHintWhenDisabledOrMissing(t *testing.T) {
|
||||||
disabled := docsTestServer(t, "")
|
disabled := docsTestServer(t, "")
|
||||||
recorder := httptest.NewRecorder()
|
recorder := httptest.NewRecorder()
|
||||||
disabled.handleUIDocs(recorder, httptest.NewRequest(http.MethodGet, "/ui/docs/", nil))
|
disabled.handleUIDocs(recorder, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/", nil))
|
||||||
if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "Documentation is not available") {
|
if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "Documentation is not available") {
|
||||||
t.Fatalf("disabled docs = %d %q", recorder.Code, recorder.Body.String())
|
t.Fatalf("disabled docs = %d %q", recorder.Code, recorder.Body.String())
|
||||||
}
|
}
|
||||||
|
|
||||||
missing := docsTestServer(t, filepath.Join(t.TempDir(), "does-not-exist"))
|
missing := docsTestServer(t, filepath.Join(t.TempDir(), "does-not-exist"))
|
||||||
recorder = httptest.NewRecorder()
|
recorder = httptest.NewRecorder()
|
||||||
missing.handleUIDocs(recorder, httptest.NewRequest(http.MethodGet, "/ui/docs/", nil))
|
missing.handleUIDocs(recorder, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/", nil))
|
||||||
if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "Documentation is not available") {
|
if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "Documentation is not available") {
|
||||||
t.Fatalf("missing docs = %d %q", recorder.Code, recorder.Body.String())
|
t.Fatalf("missing docs = %d %q", recorder.Code, recorder.Body.String())
|
||||||
}
|
}
|
||||||
@@ -82,7 +83,7 @@ func TestUIDocsShowsBuildHintWhenDisabledOrMissing(t *testing.T) {
|
|||||||
func TestUIDocsIndexRedirectsToTrailingSlash(t *testing.T) {
|
func TestUIDocsIndexRedirectsToTrailingSlash(t *testing.T) {
|
||||||
server := docsTestServer(t, t.TempDir())
|
server := docsTestServer(t, t.TempDir())
|
||||||
recorder := httptest.NewRecorder()
|
recorder := httptest.NewRecorder()
|
||||||
server.handleUIDocsIndex(recorder, httptest.NewRequest(http.MethodGet, "/ui/docs", nil))
|
server.handleUIDocsIndex(recorder, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs", nil))
|
||||||
if recorder.Code != http.StatusPermanentRedirect || recorder.Header().Get("Location") != "/ui/docs/" {
|
if recorder.Code != http.StatusPermanentRedirect || recorder.Header().Get("Location") != "/ui/docs/" {
|
||||||
t.Fatalf("redirect = %d %q", recorder.Code, recorder.Header().Get("Location"))
|
t.Fatalf("redirect = %d %q", recorder.Code, recorder.Header().Get("Location"))
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user