diff --git a/.github/workflows/bazel.yml b/.github/workflows/bazel.yml index 840114d0c..a8150f3fa 100644 --- a/.github/workflows/bazel.yml +++ b/.github/workflows/bazel.yml @@ -146,6 +146,7 @@ jobs: worker-init|src/compute-plane-services/worker-init|false|go-root|build-container worker-llm-credentials|src/compute-plane-services/worker-llm-credentials|false|go-root|build-container worker-task|src/compute-plane-services/worker-task|false|go-root|build-container + request-trace-uploader|src/compute-plane-services/request-trace-uploader|false|go-root|build-container worker-utils|src/compute-plane-services/worker-utils|false|go-root|build-container function-autoscaler|src/control-plane-services/function-autoscaler|false|go-root|build-container helm-reval|src/control-plane-services/helm-reval|false|go-root|build-container diff --git a/go.work.bazel b/go.work.bazel index 6215f1a87..c293feda0 100644 --- a/go.work.bazel +++ b/go.work.bazel @@ -41,6 +41,7 @@ use ( ./src/compute-plane-services/worker-init ./src/compute-plane-services/worker-llm-credentials ./src/compute-plane-services/worker-task + ./src/compute-plane-services/request-trace-uploader ./src/compute-plane-services/worker-utils ./src/invocation-plane-services/grpc-proxy ./src/invocation-plane-services/llm-api-gateway diff --git a/src/compute-plane-services/request-trace-uploader/AGENTS.md b/src/compute-plane-services/request-trace-uploader/AGENTS.md new file mode 100644 index 000000000..09beebf34 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/AGENTS.md @@ -0,0 +1,31 @@ +# AGENTS.md - request-trace-uploader + +Native Go sidecar scaffold for closed Dynamo request-trace segments. It does +not publish or delete source segments until a supported upload adapter lands. + +## Layout + +- `cmd/`: process entrypoint and OCI image target +- `internal/config/`: current sidecar contract and bounded policy parsing +- `internal/segment/`: closed trace/audit segment discovery +- `internal/health/`: liveness and readiness handlers +- `internal/upload/`: future upload-client boundary +- `internal/service/`: startup, recovery scan, and HTTP server + +## Build and test + +```bash +bazel test //src/compute-plane-services/request-trace-uploader/... +bazel build //src/compute-plane-services/request-trace-uploader/cmd:image +``` + +Run `bazel run //:gazelle` after changing Go imports or Bazel metadata. + +## Rules + +- Preserve the existing `trace` and `audit` capture-type names. +- Treat the highest indexed segment for each prefix as active. +- Do not add a release entry until an approved upload adapter exists. +- Do not add a Prometheus scrape endpoint. The later observability increment + exports logs, traces, and metrics through BYOO OTLP endpoints. +- Do not log request payloads, credentials, paths, or remote upload IDs. diff --git a/src/compute-plane-services/request-trace-uploader/BUILD.bazel b/src/compute-plane-services/request-trace-uploader/BUILD.bazel new file mode 100644 index 000000000..959243c17 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/BUILD.bazel @@ -0,0 +1,5 @@ +load("@gazelle//:def.bzl", "gazelle") + +# gazelle:prefix github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader +# gazelle:go_naming_convention import_alias +gazelle(name = "gazelle") diff --git a/src/compute-plane-services/request-trace-uploader/CLAUDE.md b/src/compute-plane-services/request-trace-uploader/CLAUDE.md new file mode 100644 index 000000000..43c994c2d --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/CLAUDE.md @@ -0,0 +1 @@ +@AGENTS.md diff --git a/src/compute-plane-services/request-trace-uploader/README.md b/src/compute-plane-services/request-trace-uploader/README.md new file mode 100644 index 000000000..373750242 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/README.md @@ -0,0 +1,52 @@ +# request-trace-uploader + +`request-trace-uploader` is the NVCF sidecar for Dynamo request tracing. +Dynamo calls the captured objects `RequestTraceRecord` values. The records that +contain input and output payloads have event type `request_payload`. + +The current deployment writes request-trace segments to rotating `.jsonl.gz` +files. It uses two capture types, `trace` and `audit`. This service discovers +only closed segments: the highest indexed segment for each prefix remains owned +by the Dynamo writer. + +## Initial scaffold + +This initial implementation validates the sidecar configuration, verifies its +secret-file mount, creates state and quarantine directories, exposes health, +and verifies local segment discovery. It intentionally does not export logs, +traces, or metrics, transform records, submit uploads, poll remote status, +delete source files, or publish a release image. + +The `internal/upload` package defines the future upload-client boundary. The +real adapter and durable journal are separate follow-up work. + +## Configuration + +The scaffold retains the current file contract: + +- `TRACE_DIR`: absolute directory containing trace and audit segments +- `TRACE_FILE_PREFIX`: trace segment prefix +- `AUDIT_FILE_PREFIX`: audit segment prefix +- `REQUEST_TRACE_UPLOADER_DROP_NCA_IDS`: optional CSV NCA ID drop list for audit + payloads. Bare IDs and `nca--nca` are equivalent. The future + transform retains correlation metadata and the normalized NCA ID, but removes + request and response payloads plus non-NCA headers before upload. +- `REQUEST_TRACE_UPLOADER_SECRETS_FILE`: readable mounted secret file; default + `/var/secrets/secrets.json` + +It also accepts these bounded operational settings: + +- `HEALTH_ADDR`: default `:8011` +- `UPLOAD_INTERVAL_SECONDS`: default `30` +- `STATUS_INTERVAL_SECONDS`: default `5` +- `STATUS_TIMEOUT_SECONDS`: default `900` +- `REQUEST_TRACE_UPLOADER_ATTEMPT_TIMEOUT`: default `30s` +- `REQUEST_TRACE_UPLOADER_OPERATION_TIMEOUT`: default `90s` +- `REQUEST_TRACE_UPLOADER_MAX_RETRIES`: default `2` +- `REQUEST_TRACE_UPLOADER_RETRY_INITIAL_BACKOFF`: default `100ms` +- `REQUEST_TRACE_UPLOADER_RETRY_MAX_BACKOFF`: default `15s` +- `REQUEST_TRACE_UPLOADER_RETRY_MULTIPLIER`: default `2.0` + +Invalid policy values fall back to defaults and produce a safe startup warning. +Missing paths, unreadable secret files, and incompatible required values prevent +readiness. diff --git a/src/compute-plane-services/request-trace-uploader/cmd/BUILD.bazel b/src/compute-plane-services/request-trace-uploader/cmd/BUILD.bazel new file mode 100644 index 000000000..eda05a088 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/cmd/BUILD.bazel @@ -0,0 +1,27 @@ +load("@rules_go//go:def.bzl", "go_binary", "go_library") +load("//rules/oci:defs.bzl", "go_oci_image") + +go_library( + name = "cmd_lib", + srcs = ["main.go"], + importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/cmd", + visibility = ["//visibility:private"], + deps = [ + "//src/compute-plane-services/request-trace-uploader/internal/config", + "//src/compute-plane-services/request-trace-uploader/internal/service", + ], +) + +go_binary( + name = "request-trace-uploader", + embed = [":cmd_lib"], + visibility = ["//visibility:public"], +) + +go_oci_image( + name = "image", + base = "@distroless_go", + binary = ":request-trace-uploader", + tags = ["nvcf-request-trace-uploader"], + visibility = ["//visibility:public"], +) diff --git a/src/compute-plane-services/request-trace-uploader/cmd/main.go b/src/compute-plane-services/request-trace-uploader/cmd/main.go new file mode 100644 index 000000000..fd4263c97 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/cmd/main.go @@ -0,0 +1,41 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// request-trace-uploader validates and discovers Dynamo request-trace segments. +package main + +import ( + "context" + "errors" + "log/slog" + "os" + "os/signal" + "syscall" + + "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/config" + "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/service" +) + +func main() { + cfg, warnings, err := config.LoadFromEnv() + if err != nil { + slog.Error("invalid request trace uploader configuration", "error", err) + os.Exit(1) + } + for _, warning := range warnings { + slog.Warn("request trace uploader configuration fallback", "setting", warning) + } + + svc, err := service.New(cfg) + if err != nil { + slog.Error("create request trace uploader service", "error", err) + os.Exit(1) + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := svc.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + slog.Error("request trace uploader stopped", "error", err) + os.Exit(1) + } +} diff --git a/src/compute-plane-services/request-trace-uploader/go.mod b/src/compute-plane-services/request-trace-uploader/go.mod new file mode 100644 index 000000000..e0e94dc36 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/go.mod @@ -0,0 +1,3 @@ +module github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader + +go 1.26.5 diff --git a/src/compute-plane-services/request-trace-uploader/internal/config/BUILD.bazel b/src/compute-plane-services/request-trace-uploader/internal/config/BUILD.bazel new file mode 100644 index 000000000..73bad49eb --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/config/BUILD.bazel @@ -0,0 +1,20 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "config", + srcs = ["config.go"], + importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/config", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], +) + +go_test( + name = "config_test", + srcs = ["config_test.go"], + embed = [":config"], +) + +alias( + name = "go_default_library", + actual = ":config", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], +) diff --git a/src/compute-plane-services/request-trace-uploader/internal/config/config.go b/src/compute-plane-services/request-trace-uploader/internal/config/config.go new file mode 100644 index 000000000..e86ed0c61 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/config/config.go @@ -0,0 +1,325 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// Package config loads the request-trace uploader's bounded runtime settings. +package config + +import ( + "fmt" + "math" + "os" + "path/filepath" + "strconv" + "strings" + "time" +) + +const ( + EnvTraceDir = "TRACE_DIR" + EnvTraceFilePrefix = "TRACE_FILE_PREFIX" + EnvAuditFilePrefix = "AUDIT_FILE_PREFIX" + EnvDroppedNCAIDs = "REQUEST_TRACE_UPLOADER_DROP_NCA_IDS" + EnvSecretsFile = "REQUEST_TRACE_UPLOADER_SECRETS_FILE" + EnvStateDir = "REQUEST_TRACE_UPLOADER_STATE_DIR" + EnvQuarantineDir = "REQUEST_TRACE_UPLOADER_QUARANTINE_DIR" + EnvHealthAddr = "HEALTH_ADDR" + EnvUploadInterval = "UPLOAD_INTERVAL_SECONDS" + EnvStatusInterval = "STATUS_INTERVAL_SECONDS" + EnvStatusTimeout = "STATUS_TIMEOUT_SECONDS" + EnvAttemptTimeout = "REQUEST_TRACE_UPLOADER_ATTEMPT_TIMEOUT" + EnvOperationTimeout = "REQUEST_TRACE_UPLOADER_OPERATION_TIMEOUT" + EnvMaxRetries = "REQUEST_TRACE_UPLOADER_MAX_RETRIES" + EnvRetryInitialBackoff = "REQUEST_TRACE_UPLOADER_RETRY_INITIAL_BACKOFF" + EnvRetryMaximumBackoff = "REQUEST_TRACE_UPLOADER_RETRY_MAX_BACKOFF" + EnvRetryMultiplier = "REQUEST_TRACE_UPLOADER_RETRY_MULTIPLIER" + DefaultSecretsFile = "/var/secrets/secrets.json" + DefaultHealthAddr = ":8011" + DefaultUploadInterval = 30 * time.Second + DefaultStatusInterval = 5 * time.Second + DefaultStatusTimeout = 15 * time.Minute + DefaultAttemptTimeout = 30 * time.Second + DefaultOperationTimeout = 90 * time.Second + DefaultMaxRetries = 2 + DefaultInitialBackoff = 100 * time.Millisecond + DefaultMaximumBackoff = 15 * time.Second + DefaultRetryMultiplier = 2.0 +) + +// LookupFunc obtains one environment setting. +type LookupFunc func(string) (string, bool) + +// Config is the request-trace uploader runtime configuration. +type Config struct { + TraceDir string + TraceFilePrefix string + AuditFilePrefix string + // DroppedNCAIDs identifies NCA IDs whose audit request payloads, + // response payloads, and non-NCA headers must not be exported. The current + // scaffold only validates this value. The transform stage applies it. + DroppedNCAIDs []string + SecretsFile string + StateDir string + QuarantineDir string + HealthAddr string + UploadInterval time.Duration + StatusInterval time.Duration + StatusTimeout time.Duration + RetryPolicy RetryPolicy +} + +// RetryPolicy bounds each remote operation. The initial scaffold validates but +// does not yet invoke a remote upload client. +type RetryPolicy struct { + AttemptTimeout time.Duration + OperationTimeout time.Duration + MaxRetries int + InitialBackoff time.Duration + MaximumBackoff time.Duration + Multiplier float64 +} + +// LoadFromEnv reads Config from the process environment. +func LoadFromEnv() (Config, []string, error) { + return Load(os.LookupEnv) +} + +// Load reads Config with lookup. Invalid optional policy values fall back to a +// default and add the setting name to warnings. +func Load(lookup LookupFunc) (Config, []string, error) { + if lookup == nil { + return Config{}, nil, fmt.Errorf("environment lookup is required") + } + + traceDir, err := requiredAbsolute(lookup, EnvTraceDir) + if err != nil { + return Config{}, nil, err + } + tracePrefix, err := requiredName(lookup, EnvTraceFilePrefix) + if err != nil { + return Config{}, nil, err + } + auditPrefix, err := requiredName(lookup, EnvAuditFilePrefix) + if err != nil { + return Config{}, nil, err + } + if tracePrefix == auditPrefix { + return Config{}, nil, fmt.Errorf("%s and %s must differ", EnvTraceFilePrefix, EnvAuditFilePrefix) + } + droppedNCAIDs, err := ncaIDList(lookup, EnvDroppedNCAIDs) + if err != nil { + return Config{}, nil, err + } + + warnings := make([]string, 0) + stateDir, err := optionalAbsolute(lookup, EnvStateDir, filepath.Join(traceDir, "request-trace-uploader-state")) + if err != nil { + return Config{}, nil, err + } + quarantineDir, err := optionalAbsolute(lookup, EnvQuarantineDir, filepath.Join(traceDir, "request-trace-uploader-quarantine")) + if err != nil { + return Config{}, nil, err + } + secretsFile := valueOrDefault(lookup, EnvSecretsFile, DefaultSecretsFile) + healthAddr := valueOrDefault(lookup, EnvHealthAddr, DefaultHealthAddr) + if strings.TrimSpace(healthAddr) == "" { + return Config{}, nil, fmt.Errorf("%s must not be empty", EnvHealthAddr) + } + + uploadInterval := durationSeconds(lookup, EnvUploadInterval, DefaultUploadInterval, time.Second, 24*time.Hour, &warnings) + statusInterval := durationSeconds(lookup, EnvStatusInterval, DefaultStatusInterval, time.Second, time.Hour, &warnings) + statusTimeout := durationSeconds(lookup, EnvStatusTimeout, DefaultStatusTimeout, time.Second, 24*time.Hour, &warnings) + if statusTimeout < statusInterval { + statusTimeout = statusInterval + warnings = append(warnings, EnvStatusTimeout) + } + attemptTimeout := duration(lookup, EnvAttemptTimeout, DefaultAttemptTimeout, time.Second, 90*time.Second, &warnings) + operationTimeout := duration(lookup, EnvOperationTimeout, DefaultOperationTimeout, time.Second, 5*time.Minute, &warnings) + if operationTimeout < attemptTimeout { + operationTimeout = attemptTimeout + warnings = append(warnings, EnvOperationTimeout) + } + maxRetries := integer(lookup, EnvMaxRetries, DefaultMaxRetries, 0, 10, &warnings) + initialBackoff := duration(lookup, EnvRetryInitialBackoff, DefaultInitialBackoff, 10*time.Millisecond, 10*time.Second, &warnings) + maximumBackoff := duration(lookup, EnvRetryMaximumBackoff, DefaultMaximumBackoff, 10*time.Millisecond, time.Minute, &warnings) + if maximumBackoff < initialBackoff { + maximumBackoff = initialBackoff + warnings = append(warnings, EnvRetryMaximumBackoff) + } + multiplier := floatValue(lookup, EnvRetryMultiplier, DefaultRetryMultiplier, 1.1, 10.0, &warnings) + + return Config{ + TraceDir: traceDir, + TraceFilePrefix: tracePrefix, + AuditFilePrefix: auditPrefix, + DroppedNCAIDs: droppedNCAIDs, + SecretsFile: strings.TrimSpace(secretsFile), + StateDir: stateDir, + QuarantineDir: quarantineDir, + HealthAddr: strings.TrimSpace(healthAddr), + UploadInterval: uploadInterval, + StatusInterval: statusInterval, + StatusTimeout: statusTimeout, + RetryPolicy: RetryPolicy{ + AttemptTimeout: attemptTimeout, + OperationTimeout: operationTimeout, + MaxRetries: maxRetries, + InitialBackoff: initialBackoff, + MaximumBackoff: maximumBackoff, + Multiplier: multiplier, + }, + }, warnings, nil +} + +func requiredAbsolute(lookup LookupFunc, name string) (string, error) { + value, ok := lookup(name) + if !ok || strings.TrimSpace(value) == "" { + return "", fmt.Errorf("%s is required", name) + } + return validateAbsolute(name, value) +} + +func optionalAbsolute(lookup LookupFunc, name, fallback string) (string, error) { + value, ok := lookup(name) + if !ok || strings.TrimSpace(value) == "" { + value = fallback + } + return validateAbsolute(name, value) +} + +func validateAbsolute(name, value string) (string, error) { + value = strings.TrimSpace(value) + if !filepath.IsAbs(value) { + return "", fmt.Errorf("%s must be an absolute path", name) + } + return filepath.Clean(value), nil +} + +func requiredName(lookup LookupFunc, name string) (string, error) { + value, ok := lookup(name) + value = strings.TrimSpace(value) + if !ok || value == "" { + return "", fmt.Errorf("%s is required", name) + } + if strings.ContainsAny(value, `/\\`) { + return "", fmt.Errorf("%s must not contain a path separator", name) + } + return value, nil +} + +func ncaIDList(lookup LookupFunc, name string) ([]string, error) { + value, ok := lookup(name) + if !ok || strings.TrimSpace(value) == "" { + return nil, nil + } + + ids := make([]string, 0) + seen := make(map[string]struct{}) + for _, item := range strings.Split(value, ",") { + item = strings.TrimSpace(item) + if item == "" { + continue + } + id := NormalizeNCAID(item) + if id == "" { + return nil, fmt.Errorf("%s contains an invalid NCA ID", name) + } + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + ids = append(ids, id) + } + return ids, nil +} + +// NormalizeNCAID returns the canonical form used by the payload drop list. +func NormalizeNCAID(value string) string { + value = strings.TrimSpace(value) + if strings.HasPrefix(value, "nca-") && strings.HasSuffix(value, "-nca") { + value = strings.TrimSuffix(strings.TrimPrefix(value, "nca-"), "-nca") + } + return strings.TrimSpace(value) +} + +// DropsNCAID reports whether the configured payload drop list contains value. +func (cfg Config) DropsNCAID(value string) bool { + value = NormalizeNCAID(value) + if value == "" { + return false + } + for _, id := range cfg.DroppedNCAIDs { + if id == value { + return true + } + } + return false +} + +func valueOrDefault(lookup LookupFunc, name, fallback string) string { + value, ok := lookup(name) + if !ok || strings.TrimSpace(value) == "" { + return fallback + } + return value +} + +func durationSeconds(lookup LookupFunc, name string, fallback, minimum, maximum time.Duration, warnings *[]string) time.Duration { + value, ok := lookup(name) + if !ok || strings.TrimSpace(value) == "" { + return fallback + } + seconds, err := strconv.Atoi(strings.TrimSpace(value)) + if err != nil { + *warnings = append(*warnings, name) + return fallback + } + if seconds < 0 || int64(seconds) > int64(maximum/time.Second) { + *warnings = append(*warnings, name) + return fallback + } + duration := time.Duration(seconds) * time.Second + if duration < minimum || duration > maximum { + *warnings = append(*warnings, name) + return fallback + } + return duration +} + +func duration(lookup LookupFunc, name string, fallback, minimum, maximum time.Duration, warnings *[]string) time.Duration { + value, ok := lookup(name) + if !ok || strings.TrimSpace(value) == "" { + return fallback + } + parsed, err := time.ParseDuration(strings.TrimSpace(value)) + if err != nil || parsed < minimum || parsed > maximum { + *warnings = append(*warnings, name) + return fallback + } + return parsed +} + +func integer(lookup LookupFunc, name string, fallback, minimum, maximum int, warnings *[]string) int { + value, ok := lookup(name) + if !ok || strings.TrimSpace(value) == "" { + return fallback + } + parsed, err := strconv.Atoi(strings.TrimSpace(value)) + if err != nil || parsed < minimum || parsed > maximum { + *warnings = append(*warnings, name) + return fallback + } + return parsed +} + +func floatValue(lookup LookupFunc, name string, fallback, minimum, maximum float64, warnings *[]string) float64 { + value, ok := lookup(name) + if !ok || strings.TrimSpace(value) == "" { + return fallback + } + parsed, err := strconv.ParseFloat(strings.TrimSpace(value), 64) + if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) || parsed < minimum || parsed > maximum { + *warnings = append(*warnings, name) + return fallback + } + return parsed +} diff --git a/src/compute-plane-services/request-trace-uploader/internal/config/config_test.go b/src/compute-plane-services/request-trace-uploader/internal/config/config_test.go new file mode 100644 index 000000000..2297f73c1 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/config/config_test.go @@ -0,0 +1,140 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package config + +import ( + "reflect" + "testing" + "time" +) + +func TestLoadDefaults(t *testing.T) { + cfg, warnings, err := Load(testLookup(map[string]string{ + EnvTraceDir: "/records", + EnvTraceFilePrefix: "request-trace", + EnvAuditFilePrefix: "request-audit", + })) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if len(warnings) != 0 { + t.Fatalf("warnings = %v, want none", warnings) + } + if cfg.UploadInterval != DefaultUploadInterval || cfg.StatusInterval != DefaultStatusInterval || cfg.StatusTimeout != DefaultStatusTimeout { + t.Fatalf("unexpected polling defaults: %+v", cfg) + } + if cfg.RetryPolicy.AttemptTimeout != DefaultAttemptTimeout || cfg.RetryPolicy.OperationTimeout != DefaultOperationTimeout { + t.Fatalf("unexpected retry defaults: %+v", cfg.RetryPolicy) + } + if cfg.StateDir != "/records/request-trace-uploader-state" || cfg.QuarantineDir != "/records/request-trace-uploader-quarantine" { + t.Fatalf("unexpected derived directories: state=%q quarantine=%q", cfg.StateDir, cfg.QuarantineDir) + } +} + +func TestLoadNormalizesDroppedNCAIDs(t *testing.T) { + cfg, warnings, err := Load(testLookup(map[string]string{ + EnvTraceDir: "/records", + EnvTraceFilePrefix: "request-trace", + EnvAuditFilePrefix: "request-audit", + EnvDroppedNCAIDs: " first, nca-second-nca, first, , third ", + })) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if len(warnings) != 0 { + t.Fatalf("warnings = %v, want none", warnings) + } + if want := []string{"first", "second", "third"}; !reflect.DeepEqual(cfg.DroppedNCAIDs, want) { + t.Errorf("DroppedNCAIDs = %v, want %v", cfg.DroppedNCAIDs, want) + } +} + +func TestLoadRejectsInvalidDroppedNCAID(t *testing.T) { + _, _, err := Load(testLookup(map[string]string{ + EnvTraceDir: "/records", + EnvTraceFilePrefix: "request-trace", + EnvAuditFilePrefix: "request-audit", + EnvDroppedNCAIDs: "nca--nca", + })) + if err == nil { + t.Fatal("Load() error = nil, want invalid NCA ID error") + } +} + +func TestConfigDropsNCAID(t *testing.T) { + cfg, _, err := Load(testLookup(map[string]string{ + EnvTraceDir: "/records", + EnvTraceFilePrefix: "request-trace", + EnvAuditFilePrefix: "request-audit", + EnvDroppedNCAIDs: "customer", + })) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if !cfg.DropsNCAID("nca-customer-nca") { + t.Error("DropsNCAID(wrapper) = false, want true") + } + if cfg.DropsNCAID("CUSTOMER") { + t.Error("DropsNCAID(case changed) = true, want false") + } + if cfg.DropsNCAID("") { + t.Error("DropsNCAID(empty) = true, want false") + } +} + +func TestLoadFallsBackForInvalidPolicy(t *testing.T) { + cfg, warnings, err := Load(testLookup(map[string]string{ + EnvTraceDir: "/records", + EnvTraceFilePrefix: "request-trace", + EnvAuditFilePrefix: "request-audit", + EnvAttemptTimeout: "0s", + EnvOperationTimeout: "10s", + EnvMaxRetries: "99", + EnvRetryInitialBackoff: "not-a-duration", + EnvRetryMaximumBackoff: "1ms", + EnvRetryMultiplier: "nan", + EnvStatusTimeout: "1", + EnvStatusInterval: "10", + })) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if cfg.RetryPolicy.AttemptTimeout != DefaultAttemptTimeout { + t.Errorf("attempt timeout = %v, want %v", cfg.RetryPolicy.AttemptTimeout, DefaultAttemptTimeout) + } + if cfg.RetryPolicy.MaxRetries != DefaultMaxRetries { + t.Errorf("max retries = %d, want %d", cfg.RetryPolicy.MaxRetries, DefaultMaxRetries) + } + if cfg.StatusTimeout != 10*time.Second { + t.Errorf("status timeout = %v, want clamped %v", cfg.StatusTimeout, 10*time.Second) + } + if len(warnings) < 6 { + t.Errorf("warnings = %v, want policy fallbacks", warnings) + } +} + +func TestLoadRejectsInvalidRequiredValues(t *testing.T) { + tests := []struct { + name string + env map[string]string + }{ + {name: "missing directory", env: map[string]string{EnvTraceFilePrefix: "trace", EnvAuditFilePrefix: "audit"}}, + {name: "relative directory", env: map[string]string{EnvTraceDir: "records", EnvTraceFilePrefix: "trace", EnvAuditFilePrefix: "audit"}}, + {name: "same prefix", env: map[string]string{EnvTraceDir: "/records", EnvTraceFilePrefix: "trace", EnvAuditFilePrefix: "trace"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if _, _, err := Load(testLookup(tt.env)); err == nil { + t.Fatal("Load() error = nil, want error") + } + }) + } +} + +func testLookup(values map[string]string) LookupFunc { + return func(name string) (string, bool) { + value, ok := values[name] + return value, ok + } +} diff --git a/src/compute-plane-services/request-trace-uploader/internal/health/BUILD.bazel b/src/compute-plane-services/request-trace-uploader/internal/health/BUILD.bazel new file mode 100644 index 000000000..70b88568d --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/health/BUILD.bazel @@ -0,0 +1,20 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "health", + srcs = ["health.go"], + importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/health", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], +) + +go_test( + name = "health_test", + srcs = ["health_test.go"], + embed = [":health"], +) + +alias( + name = "go_default_library", + actual = ":health", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], +) diff --git a/src/compute-plane-services/request-trace-uploader/internal/health/health.go b/src/compute-plane-services/request-trace-uploader/internal/health/health.go new file mode 100644 index 000000000..feedf72e9 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/health/health.go @@ -0,0 +1,40 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// Package health exposes liveness and readiness endpoints. +package health + +import ( + "net/http" + "sync/atomic" +) + +// Handler exposes liveness and readiness for the uploader process. +type Handler struct { + ready atomic.Bool +} + +// New returns an unready Handler. The running process is always live. +func New() *Handler { + return &Handler{} +} + +// SetReady updates readiness after local startup checks complete. +func (h *Handler) SetReady(ready bool) { + h.ready.Store(ready) +} + +// Live handles the liveness endpoint. +func (h *Handler) Live(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) +} + +// Ready handles the readiness endpoint. It intentionally does not depend on +// a remote destination or the current backlog. +func (h *Handler) Ready(w http.ResponseWriter, _ *http.Request) { + if !h.ready.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) +} diff --git a/src/compute-plane-services/request-trace-uploader/internal/health/health_test.go b/src/compute-plane-services/request-trace-uploader/internal/health/health_test.go new file mode 100644 index 000000000..b6effd266 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/health/health_test.go @@ -0,0 +1,31 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package health + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +func TestEndpoints(t *testing.T) { + h := New() + live := httptest.NewRecorder() + h.Live(live, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/livez", nil)) + if live.Code != http.StatusOK { + t.Fatalf("live status = %d, want %d", live.Code, http.StatusOK) + } + ready := httptest.NewRecorder() + h.Ready(ready, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/readyz", nil)) + if ready.Code != http.StatusServiceUnavailable { + t.Fatalf("initial ready status = %d, want %d", ready.Code, http.StatusServiceUnavailable) + } + h.SetReady(true) + ready = httptest.NewRecorder() + h.Ready(ready, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/readyz", nil)) + if ready.Code != http.StatusOK { + t.Fatalf("ready status = %d, want %d", ready.Code, http.StatusOK) + } +} diff --git a/src/compute-plane-services/request-trace-uploader/internal/segment/BUILD.bazel b/src/compute-plane-services/request-trace-uploader/internal/segment/BUILD.bazel new file mode 100644 index 000000000..1d157d7ad --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/segment/BUILD.bazel @@ -0,0 +1,20 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "segment", + srcs = ["segment.go"], + importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/segment", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], +) + +go_test( + name = "segment_test", + srcs = ["segment_test.go"], + embed = [":segment"], +) + +alias( + name = "go_default_library", + actual = ":segment", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], +) diff --git a/src/compute-plane-services/request-trace-uploader/internal/segment/segment.go b/src/compute-plane-services/request-trace-uploader/internal/segment/segment.go new file mode 100644 index 000000000..db42c9e81 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/segment/segment.go @@ -0,0 +1,92 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// Package segment discovers closed Dynamo request-trace segments. +package segment + +import ( + "fmt" + "os" + "path/filepath" + "regexp" + "sort" + "strconv" + "time" +) + +// CaptureType identifies the current Dynamo deployment's two trace streams. +type CaptureType string + +const ( + CaptureTypeTrace CaptureType = "trace" + CaptureTypeAudit CaptureType = "audit" +) + +// Segment is one closed, compressed Dynamo request-trace segment. +type Segment struct { + CaptureType CaptureType + Path string + Index int + Size int64 + ModTime time.Time +} + +// Discover returns segments that are safe for a future uploader to process. +// Dynamo appends gzip members to the highest indexed segment for each prefix, +// so the scanner always leaves that segment untouched. +func Discover(directory, tracePrefix, auditPrefix string) ([]Segment, error) { + trace, err := discoverPrefix(directory, tracePrefix, CaptureTypeTrace) + if err != nil { + return nil, err + } + audit, err := discoverPrefix(directory, auditPrefix, CaptureTypeAudit) + if err != nil { + return nil, err + } + segments := append(trace, audit...) + sort.Slice(segments, func(i, j int) bool { + if segments[i].ModTime.Equal(segments[j].ModTime) { + return segments[i].Path < segments[j].Path + } + return segments[i].ModTime.Before(segments[j].ModTime) + }) + return segments, nil +} + +func discoverPrefix(directory, prefix string, captureType CaptureType) ([]Segment, error) { + entries, err := os.ReadDir(directory) + if err != nil { + return nil, fmt.Errorf("read request trace directory: %w", err) + } + pattern := regexp.MustCompile("^" + regexp.QuoteMeta(prefix) + `\.(\d{6})\.jsonl\.gz$`) + segments := make([]Segment, 0) + for _, entry := range entries { + if !entry.Type().IsRegular() { + continue + } + matches := pattern.FindStringSubmatch(entry.Name()) + if matches == nil { + continue + } + index, err := strconv.Atoi(matches[1]) + if err != nil { + return nil, fmt.Errorf("parse request trace segment index %q: %w", matches[1], err) + } + info, err := entry.Info() + if err != nil { + return nil, fmt.Errorf("stat request trace segment %q: %w", entry.Name(), err) + } + segments = append(segments, Segment{ + CaptureType: captureType, + Path: filepath.Join(directory, entry.Name()), + Index: index, + Size: info.Size(), + ModTime: info.ModTime(), + }) + } + sort.Slice(segments, func(i, j int) bool { return segments[i].Index < segments[j].Index }) + if len(segments) < 2 { + return nil, nil + } + return segments[:len(segments)-1], nil +} diff --git a/src/compute-plane-services/request-trace-uploader/internal/segment/segment_test.go b/src/compute-plane-services/request-trace-uploader/internal/segment/segment_test.go new file mode 100644 index 000000000..d421805c9 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/segment/segment_test.go @@ -0,0 +1,54 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package segment + +import ( + "os" + "path/filepath" + "testing" +) + +func TestDiscoverExcludesActiveSegmentForEachCaptureType(t *testing.T) { + directory := t.TempDir() + for _, name := range []string{ + "request-trace.000000.jsonl.gz", + "request-trace.000001.jsonl.gz", + "request-audit.000007.jsonl.gz", + "request-audit.000008.jsonl.gz", + "unrelated.jsonl.gz", + } { + if err := os.WriteFile(filepath.Join(directory, name), []byte("fixture"), 0o600); err != nil { + t.Fatalf("WriteFile(%q): %v", name, err) + } + } + + segments, err := Discover(directory, "request-trace", "request-audit") + if err != nil { + t.Fatalf("Discover() error = %v", err) + } + if len(segments) != 2 { + t.Fatalf("segments = %d, want 2: %#v", len(segments), segments) + } + got := map[CaptureType]int{} + for _, item := range segments { + got[item.CaptureType] = item.Index + } + if got[CaptureTypeTrace] != 0 || got[CaptureTypeAudit] != 7 { + t.Fatalf("closed indexes = %#v, want trace=0 audit=7", got) + } +} + +func TestDiscoverLeavesOnlySegmentActive(t *testing.T) { + directory := t.TempDir() + if err := os.WriteFile(filepath.Join(directory, "request-trace.000000.jsonl.gz"), []byte("fixture"), 0o600); err != nil { + t.Fatal(err) + } + segments, err := Discover(directory, "request-trace", "request-audit") + if err != nil { + t.Fatalf("Discover() error = %v", err) + } + if len(segments) != 0 { + t.Fatalf("segments = %#v, want none", segments) + } +} diff --git a/src/compute-plane-services/request-trace-uploader/internal/service/BUILD.bazel b/src/compute-plane-services/request-trace-uploader/internal/service/BUILD.bazel new file mode 100644 index 000000000..6059ce16c --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/service/BUILD.bazel @@ -0,0 +1,26 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "service", + srcs = ["service.go"], + importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/service", + visibility = ["//visibility:public"], + deps = [ + "//src/compute-plane-services/request-trace-uploader/internal/config", + "//src/compute-plane-services/request-trace-uploader/internal/health", + "//src/compute-plane-services/request-trace-uploader/internal/segment", + ], +) + +go_test( + name = "service_test", + srcs = ["service_test.go"], + embed = [":service"], + deps = ["//src/compute-plane-services/request-trace-uploader/internal/config"], +) + +alias( + name = "go_default_library", + actual = ":service", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], +) diff --git a/src/compute-plane-services/request-trace-uploader/internal/service/service.go b/src/compute-plane-services/request-trace-uploader/internal/service/service.go new file mode 100644 index 000000000..c8974093b --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/service/service.go @@ -0,0 +1,117 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// Package service starts the safe request-trace uploader scaffold. +package service + +import ( + "context" + "errors" + "fmt" + "net/http" + "os" + "time" + + "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/config" + "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/health" + "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/segment" +) + +// Service owns local readiness checks and the sidecar HTTP server. It +// intentionally does not submit or delete request-trace segments. +type Service struct { + config config.Config + health *health.Handler +} + +// New creates a request-trace uploader service. +func New(cfg config.Config) (*Service, error) { + return &Service{ + config: cfg, + health: health.New(), + }, nil +} + +// Handler returns the service HTTP handler. +func (s *Service) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("GET /livez", s.health.Live) + mux.HandleFunc("GET /readyz", s.health.Ready) + return mux +} + +// Initialize performs local, non-destructive startup checks. Remote +// reachability and backlog state do not affect readiness. +func (s *Service) Initialize() error { + for _, directory := range []string{s.config.StateDir, s.config.QuarantineDir} { + if err := os.MkdirAll(directory, 0o750); err != nil { + return fmt.Errorf("create uploader directory: %w", err) + } + } + secret, err := os.Open(s.config.SecretsFile) + if err != nil { + return fmt.Errorf("open uploader secret file: %w", err) + } + if err := secret.Close(); err != nil { + return fmt.Errorf("close uploader secret file: %w", err) + } + if err := s.Refresh(); err != nil { + return fmt.Errorf("refresh local segment state: %w", err) + } + s.health.SetReady(true) + return nil +} + +// Refresh verifies that local request-trace segment discovery succeeds without +// changing source files. +func (s *Service) Refresh() error { + if _, err := segment.Discover(s.config.TraceDir, s.config.TraceFilePrefix, s.config.AuditFilePrefix); err != nil { + return fmt.Errorf("discover request trace segments: %w", err) + } + return nil +} + +// Run starts the HTTP server and periodically refreshes local discovery. +func (s *Service) Run(ctx context.Context) error { + if err := s.Initialize(); err != nil { + return fmt.Errorf("initialize request-trace uploader: %w", err) + } + server := s.httpServer() + errs := make(chan error, 1) + go func() { + if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + errs <- err + } + }() + + ticker := time.NewTicker(s.config.UploadInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + return fmt.Errorf("shutdown uploader HTTP server: %w", err) + } + return ctx.Err() + case err := <-errs: + return fmt.Errorf("serve uploader HTTP endpoints: %w", err) + case <-ticker.C: + if err := s.Refresh(); err != nil { + return fmt.Errorf("refresh request trace segments: %w", err) + } + } + } +} + +func (s *Service) httpServer() *http.Server { + return &http.Server{ + Addr: s.config.HealthAddr, + Handler: s.Handler(), + ReadHeaderTimeout: 5 * time.Second, + ReadTimeout: 15 * time.Second, + WriteTimeout: 15 * time.Second, + IdleTimeout: 60 * time.Second, + } +} diff --git a/src/compute-plane-services/request-trace-uploader/internal/service/service_test.go b/src/compute-plane-services/request-trace-uploader/internal/service/service_test.go new file mode 100644 index 000000000..fcdcab9a5 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/service/service_test.go @@ -0,0 +1,102 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package service + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/config" +) + +func TestInitializeReadinessAndDiscovery(t *testing.T) { + root := t.TempDir() + secretsFile := filepath.Join(root, "secrets.json") + if err := os.WriteFile(secretsFile, []byte("{}"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(root, "request-trace.000000.jsonl.gz"), []byte("closed"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(root, "request-trace.000001.jsonl.gz"), []byte("active"), 0o600); err != nil { + t.Fatal(err) + } + cfg := config.Config{ + TraceDir: root, + TraceFilePrefix: "request-trace", + AuditFilePrefix: "request-audit", + SecretsFile: secretsFile, + StateDir: filepath.Join(root, "state"), + QuarantineDir: filepath.Join(root, "quarantine"), + HealthAddr: ":8011", + UploadInterval: config.DefaultUploadInterval, + } + svc, err := New(cfg) + if err != nil { + t.Fatalf("New() error = %v", err) + } + if err := svc.Initialize(); err != nil { + t.Fatalf("Initialize() error = %v", err) + } + for path, want := range map[string]int{ + "/livez": http.StatusOK, + "/readyz": http.StatusOK, + "/metrics": http.StatusNotFound, + } { + response := httptest.NewRecorder() + svc.Handler().ServeHTTP(response, httptest.NewRequestWithContext(context.Background(), http.MethodGet, path, nil)) + if response.Code != want { + t.Errorf("%s status = %d, want %d", path, response.Code, want) + } + } + if _, err := os.Stat(cfg.StateDir); err != nil { + t.Errorf("state directory: %v", err) + } + if _, err := os.Stat(cfg.QuarantineDir); err != nil { + t.Errorf("quarantine directory: %v", err) + } +} + +func TestHTTPServerTimeouts(t *testing.T) { + svc, err := New(config.Config{HealthAddr: ":8011"}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + server := svc.httpServer() + if server.ReadHeaderTimeout != 5*time.Second { + t.Errorf("ReadHeaderTimeout = %v, want %v", server.ReadHeaderTimeout, 5*time.Second) + } + if server.ReadTimeout != 15*time.Second { + t.Errorf("ReadTimeout = %v, want %v", server.ReadTimeout, 15*time.Second) + } + if server.WriteTimeout != 15*time.Second { + t.Errorf("WriteTimeout = %v, want %v", server.WriteTimeout, 15*time.Second) + } + if server.IdleTimeout != 60*time.Second { + t.Errorf("IdleTimeout = %v, want %v", server.IdleTimeout, 60*time.Second) + } +} + +func TestInitializeRejectsUnreadableSecret(t *testing.T) { + root := t.TempDir() + svc, err := New(config.Config{ + TraceDir: root, + TraceFilePrefix: "request-trace", + AuditFilePrefix: "request-audit", + SecretsFile: filepath.Join(root, "missing.json"), + StateDir: filepath.Join(root, "state"), + QuarantineDir: filepath.Join(root, "quarantine"), + }) + if err != nil { + t.Fatalf("New() error = %v", err) + } + if err := svc.Initialize(); err == nil { + t.Fatal("Initialize() error = nil, want error") + } +} diff --git a/src/compute-plane-services/request-trace-uploader/internal/upload/BUILD.bazel b/src/compute-plane-services/request-trace-uploader/internal/upload/BUILD.bazel new file mode 100644 index 000000000..93cd6a9b9 --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/upload/BUILD.bazel @@ -0,0 +1,15 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "upload", + srcs = ["client.go"], + importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/upload", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], + deps = ["//src/compute-plane-services/request-trace-uploader/internal/segment"], +) + +alias( + name = "go_default_library", + actual = ":upload", + visibility = ["//src/compute-plane-services/request-trace-uploader:__subpackages__"], +) diff --git a/src/compute-plane-services/request-trace-uploader/internal/upload/client.go b/src/compute-plane-services/request-trace-uploader/internal/upload/client.go new file mode 100644 index 000000000..a29f3c67a --- /dev/null +++ b/src/compute-plane-services/request-trace-uploader/internal/upload/client.go @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// Package upload defines the future request-trace upload boundary. +package upload + +import ( + "context" + + "github.com/NVIDIA/nvcf/src/compute-plane-services/request-trace-uploader/internal/segment" +) + +// Client submits one prepared trace segment and reads its terminal status. +// The initial scaffold intentionally does not provide an implementation. +type Client interface { + Submit(context.Context, SubmitRequest) (string, error) + Status(context.Context, string) (Status, error) +} + +// SubmitRequest identifies one prepared segment without exposing its contents. +type SubmitRequest struct { + Segment segment.Segment + Path string +} + +// Status is an upload operation state. +type Status string + +const ( + StatusPending Status = "pending" + StatusSuccess Status = "success" + StatusFailure Status = "failure" +)