From e63fb7d59c0a078baed50e0c2dac209eb1851e6b Mon Sep 17 00:00:00 2001 From: Christopher Maher Date: Tue, 14 Jul 2026 10:49:24 -0700 Subject: [PATCH] feat(controller): add s3:// model source via curl --aws-sigv4 (#1098) Support s3:// (S3-compatible: MinIO, etc.) as a Model.spec.source, fetched by the model-downloader init container using curl SigV4. Additive to the existing string source; no CRD refactor. - isS3Source / parseS3Source in source.go; isHFRepoSource rejects s3:// so an s3 URL is never misclassified as a HuggingFace repo. - ModelSpec.SourceSecretRef (corev1.LocalObjectReference) supplies the S3-compatible credentials/endpoint/region via a Secret, wired as EnvFrom on the init container (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION, AWS_ENDPOINT_URL). - Source validation Pattern widened to allow the s3 scheme. - buildModelInitCommand emits a curl --aws-sigv4 path-style download with the same idempotency wrapper as the http branch; modelInitEnvVars adds S3_BUCKET / S3_KEY. - config/samples/model_s3_source.yaml (Secret + Model). - Unit tests for parse, classification, init command, and env wiring. Full end-to-end pull requires a live S3/MinIO and is validated manually; the parse plus command/env construction is the in-workspace-provable slice. Fixes #1098 Signed-off-by: Christopher Maher --- api/v1alpha1/model_types.go | 13 ++- api/v1alpha1/zz_generated.deepcopy.go | 5 + charts/llmkube/templates/crds/models.yaml | 24 ++++- .../bases/inference.llmkube.dev_models.yaml | 24 ++++- config/samples/model_s3_source.yaml | 31 ++++++ .../inferenceservice_storage_test.go | 22 ++--- internal/controller/model_storage.go | 40 +++++++- internal/controller/model_storage_test.go | 96 +++++++++++++++++++ internal/controller/source.go | 43 +++++++++ internal/controller/source_test.go | 76 +++++++++++++++ 10 files changed, 353 insertions(+), 21 deletions(-) create mode 100644 config/samples/model_s3_source.yaml create mode 100644 internal/controller/model_storage_test.go diff --git a/api/v1alpha1/model_types.go b/api/v1alpha1/model_types.go index e461e0b5..45da8b79 100644 --- a/api/v1alpha1/model_types.go +++ b/api/v1alpha1/model_types.go @@ -26,12 +26,13 @@ type ModelSpec struct { // Source defines where to obtain the model. // For GGUF models: URL or path to a .gguf file. // For MLX models: local directory path containing the model (config.json, weights). - // Supported schemes: http://, https://, file://, pvc://, or absolute paths. + // Supported schemes: http://, https://, file://, pvc://, hf://, s3://, or absolute paths. // Examples: // - https://huggingface.co/org/repo/resolve/main/model.gguf // - file:///mnt/models/model.gguf // - /mnt/models/model.gguf (air-gapped deployments) // - pvc://my-models-pvc/path/to/model.gguf (pre-staged on a PersistentVolumeClaim) + // - s3://my-bucket/models/llama-3.1-8b-q4_k_m.gguf (S3-compatible object store) // - /mnt/models/Llama-3.2-3B-Instruct-4bit (MLX model directory) // // file:// caveat for hybrid topologies: the controller pod must be @@ -44,7 +45,7 @@ type ModelSpec struct { // equivalent https://huggingface.co/.../.gguf URL which // the runtime/init container resolves at deploy time. // +kubebuilder:validation:Required - // +kubebuilder:validation:Pattern=`^(https?|file|pvc|hf)://.*|^/[^\s]+$|^[a-zA-Z0-9][\w\-\.\/]+$` + // +kubebuilder:validation:Pattern=`^(https?|file|pvc|hf|s3)://.*|^/[^\s]+$|^[a-zA-Z0-9][\w\-\.\/]+$` Source string `json:"source"` // SHA256 is the expected SHA256 hash of the model file for integrity verification. @@ -53,6 +54,14 @@ type ModelSpec struct { // +optional SHA256 string `json:"sha256,omitempty"` + // SourceSecretRef names a Secret (in the Model's namespace) whose keys are + // wired as env into the model-downloader init container. Used by s3:// + // sources for S3-compatible credentials/endpoint: AWS_ACCESS_KEY_ID, + // AWS_SECRET_ACCESS_KEY, AWS_REGION, AWS_ENDPOINT_URL (path-style, e.g. + // https://minio.internal:9000 or https://s3.us-east-1.amazonaws.com). + // +optional + SourceSecretRef *corev1.LocalObjectReference `json:"sourceSecretRef,omitempty"` + // RefreshPolicy controls whether a cached model file is re-fetched when the // upstream source changes. // diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index b7e25e63..8a5b7f45 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -1243,6 +1243,11 @@ func (in *ModelRouterStatus) DeepCopy() *ModelRouterStatus { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ModelSpec) DeepCopyInto(out *ModelSpec) { *out = *in + if in.SourceSecretRef != nil { + in, out := &in.SourceSecretRef, &out.SourceSecretRef + *out = new(corev1.LocalObjectReference) + **out = **in + } if in.Hardware != nil { in, out := &in.Hardware, &out.Hardware *out = new(HardwareSpec) diff --git a/charts/llmkube/templates/crds/models.yaml b/charts/llmkube/templates/crds/models.yaml index 5a73f92f..2c899961 100644 --- a/charts/llmkube/templates/crds/models.yaml +++ b/charts/llmkube/templates/crds/models.yaml @@ -357,12 +357,13 @@ spec: Source defines where to obtain the model. For GGUF models: URL or path to a .gguf file. For MLX models: local directory path containing the model (config.json, weights). - Supported schemes: http://, https://, file://, pvc://, or absolute paths. + Supported schemes: http://, https://, file://, pvc://, hf://, s3://, or absolute paths. Examples: - https://huggingface.co/org/repo/resolve/main/model.gguf - file:///mnt/models/model.gguf - /mnt/models/model.gguf (air-gapped deployments) - pvc://my-models-pvc/path/to/model.gguf (pre-staged on a PersistentVolumeClaim) + - s3://my-bucket/models/llama-3.1-8b-q4_k_m.gguf (S3-compatible object store) - /mnt/models/Llama-3.2-3B-Instruct-4bit (MLX model directory) file:// caveat for hybrid topologies: the controller pod must be @@ -374,8 +375,27 @@ spec: tightly (#405). Workaround: pre-stage on a pvc://, or use the equivalent https://huggingface.co/.../.gguf URL which the runtime/init container resolves at deploy time. - pattern: ^(https?|file|pvc|hf)://.*|^/[^\s]+$|^[a-zA-Z0-9][\w\-\.\/]+$ + pattern: ^(https?|file|pvc|hf|s3)://.*|^/[^\s]+$|^[a-zA-Z0-9][\w\-\.\/]+$ type: string + sourceSecretRef: + description: |- + SourceSecretRef names a Secret (in the Model's namespace) whose keys are + wired as env into the model-downloader init container. Used by s3:// + sources for S3-compatible credentials/endpoint: AWS_ACCESS_KEY_ID, + AWS_SECRET_ACCESS_KEY, AWS_REGION, AWS_ENDPOINT_URL (path-style, e.g. + https://minio.internal:9000 or https://s3.us-east-1.amazonaws.com). + properties: + name: + default: "" + description: |- + Name of the referent. + This field is effectively required, but due to backwards compatibility is + allowed to be empty. Instances of this type with an empty value here are + almost certainly wrong. + More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names + type: string + type: object + x-kubernetes-map-type: atomic required: - source type: object diff --git a/config/crd/bases/inference.llmkube.dev_models.yaml b/config/crd/bases/inference.llmkube.dev_models.yaml index 50e1172f..1c8eaae7 100644 --- a/config/crd/bases/inference.llmkube.dev_models.yaml +++ b/config/crd/bases/inference.llmkube.dev_models.yaml @@ -353,12 +353,13 @@ spec: Source defines where to obtain the model. For GGUF models: URL or path to a .gguf file. For MLX models: local directory path containing the model (config.json, weights). - Supported schemes: http://, https://, file://, pvc://, or absolute paths. + Supported schemes: http://, https://, file://, pvc://, hf://, s3://, or absolute paths. Examples: - https://huggingface.co/org/repo/resolve/main/model.gguf - file:///mnt/models/model.gguf - /mnt/models/model.gguf (air-gapped deployments) - pvc://my-models-pvc/path/to/model.gguf (pre-staged on a PersistentVolumeClaim) + - s3://my-bucket/models/llama-3.1-8b-q4_k_m.gguf (S3-compatible object store) - /mnt/models/Llama-3.2-3B-Instruct-4bit (MLX model directory) file:// caveat for hybrid topologies: the controller pod must be @@ -370,8 +371,27 @@ spec: tightly (#405). Workaround: pre-stage on a pvc://, or use the equivalent https://huggingface.co/.../.gguf URL which the runtime/init container resolves at deploy time. - pattern: ^(https?|file|pvc|hf)://.*|^/[^\s]+$|^[a-zA-Z0-9][\w\-\.\/]+$ + pattern: ^(https?|file|pvc|hf|s3)://.*|^/[^\s]+$|^[a-zA-Z0-9][\w\-\.\/]+$ type: string + sourceSecretRef: + description: |- + SourceSecretRef names a Secret (in the Model's namespace) whose keys are + wired as env into the model-downloader init container. Used by s3:// + sources for S3-compatible credentials/endpoint: AWS_ACCESS_KEY_ID, + AWS_SECRET_ACCESS_KEY, AWS_REGION, AWS_ENDPOINT_URL (path-style, e.g. + https://minio.internal:9000 or https://s3.us-east-1.amazonaws.com). + properties: + name: + default: "" + description: |- + Name of the referent. + This field is effectively required, but due to backwards compatibility is + allowed to be empty. Instances of this type with an empty value here are + almost certainly wrong. + More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names + type: string + type: object + x-kubernetes-map-type: atomic required: - source type: object diff --git a/config/samples/model_s3_source.yaml b/config/samples/model_s3_source.yaml new file mode 100644 index 00000000..cc14fae6 --- /dev/null +++ b/config/samples/model_s3_source.yaml @@ -0,0 +1,31 @@ +# Example Model with an S3 source. The init container fetches the model +# using curl --aws-sigv4 with credentials and endpoint from the Secret +# referenced by spec.sourceSecretRef. +# +# Full pull requires a live S3-compatible endpoint (e.g. MinIO) and is +# validated manually. +apiVersion: v1 +kind: Secret +metadata: + name: s3-credentials + namespace: default +stringData: + AWS_ACCESS_KEY_ID: "minioadmin" + AWS_SECRET_ACCESS_KEY: "minioadmin" + AWS_REGION: "us-east-1" + AWS_ENDPOINT_URL: "https://minio.internal:9000" +--- +apiVersion: inference.llmkube.dev/v1alpha1 +kind: Model +metadata: + labels: + app.kubernetes.io/name: llmkube + app.kubernetes.io/managed-by: kustomize + name: llama-3.1-8b-s3 + namespace: default +spec: + source: s3://my-bucket/models/llama-3.1-8b-q4_k_m.gguf + sourceSecretRef: + name: s3-credentials + format: gguf + quantization: Q4_K_M diff --git a/internal/controller/inferenceservice_storage_test.go b/internal/controller/inferenceservice_storage_test.go index a026b406..3019b869 100644 --- a/internal/controller/inferenceservice_storage_test.go +++ b/internal/controller/inferenceservice_storage_test.go @@ -1183,7 +1183,7 @@ var _ = Describe("resolveCacheMode", func() { var _ = Describe("buildModelInitCommand", func() { It("should generate cached remote download command with env var references", func() { - cmd := buildModelInitCommand(false, true, RefreshPolicyIfNotPresent) + cmd := buildModelInitCommand(false, false, true, RefreshPolicyIfNotPresent) Expect(cmd).To(ContainSubstring(`mkdir -p "$CACHE_DIR"`)) Expect(cmd).To(ContainSubstring(`"$MODEL_PATH"`)) Expect(cmd).To(ContainSubstring("curl -f -L")) @@ -1191,20 +1191,20 @@ var _ = Describe("buildModelInitCommand", func() { }) It("should generate cached local copy command", func() { - cmd := buildModelInitCommand(true, true, RefreshPolicyIfNotPresent) + cmd := buildModelInitCommand(true, false, true, RefreshPolicyIfNotPresent) Expect(cmd).To(ContainSubstring(`mkdir -p "$CACHE_DIR"`)) Expect(cmd).To(ContainSubstring("cp /host-model/model.gguf")) Expect(cmd).To(ContainSubstring(`"$MODEL_PATH"`)) }) It("should generate error exit for uncached local source", func() { - cmd := buildModelInitCommand(true, false, RefreshPolicyIfNotPresent) + cmd := buildModelInitCommand(true, false, false, RefreshPolicyIfNotPresent) Expect(cmd).To(ContainSubstring("ERROR: Local model source requires model cache")) Expect(cmd).To(ContainSubstring("exit 1")) }) It("should generate uncached remote download command with env var references", func() { - cmd := buildModelInitCommand(false, false, RefreshPolicyIfNotPresent) + cmd := buildModelInitCommand(false, false, false, RefreshPolicyIfNotPresent) Expect(cmd).To(ContainSubstring("curl -f -L")) Expect(cmd).To(ContainSubstring(`"$MODEL_SOURCE"`)) Expect(cmd).To(ContainSubstring(`"$MODEL_PATH"`)) @@ -1215,7 +1215,7 @@ var _ = Describe("buildModelInitCommand", func() { // Verify that a malicious source cannot appear in the shell script. // The command is a static template with env var references only. maliciousSource := `https://evil.com/$(touch /pwned).gguf` - cmd := buildModelInitCommand(false, true, RefreshPolicyIfNotPresent) + cmd := buildModelInitCommand(false, false, true, RefreshPolicyIfNotPresent) Expect(cmd).NotTo(ContainSubstring(maliciousSource)) Expect(cmd).NotTo(ContainSubstring("touch")) Expect(cmd).NotTo(ContainSubstring("evil.com")) @@ -1228,7 +1228,7 @@ var _ = Describe("buildModelInitCommand", func() { Context("RefreshPolicy=OnChange (http/https revalidation, issue #619)", func() { It("cached: emits curl conditional GET against an etag marker beside the model", func() { - cmd := buildModelInitCommand(false, true, RefreshPolicyOnChange) + cmd := buildModelInitCommand(false, false, true, RefreshPolicyOnChange) // Still provisions the cache dir like IfNotPresent. Expect(cmd).To(ContainSubstring(`mkdir -p "$CACHE_DIR"`)) // Conditional GET via curl's native ETag flags. @@ -1243,7 +1243,7 @@ var _ = Describe("buildModelInitCommand", func() { }) It("uncached: emits the same conditional GET without the cache dir mkdir", func() { - cmd := buildModelInitCommand(false, false, RefreshPolicyOnChange) + cmd := buildModelInitCommand(false, false, false, RefreshPolicyOnChange) Expect(cmd).To(ContainSubstring("--etag-compare")) Expect(cmd).To(ContainSubstring("--etag-save")) Expect(cmd).To(ContainSubstring(`"$MODEL_SOURCE"`)) @@ -1252,7 +1252,7 @@ var _ = Describe("buildModelInitCommand", func() { }) It("keeps the cached file and exits 0 when revalidation is unreachable", func() { - cmd := buildModelInitCommand(false, true, RefreshPolicyOnChange) + cmd := buildModelInitCommand(false, false, true, RefreshPolicyOnChange) // Robustness guard: a network blip must not take down a running // InferenceService on pod restart. Expect(cmd).To(ContainSubstring(`[ -f "$MODEL_PATH" ]`)) @@ -1265,14 +1265,14 @@ var _ = Describe("buildModelInitCommand", func() { It("does not change the local (file://) init path", func() { // file:// sources are owned by the controller (#635); the init // container path must be identical regardless of RefreshPolicy. - ifNotPresent := buildModelInitCommand(true, true, RefreshPolicyIfNotPresent) - onChange := buildModelInitCommand(true, true, RefreshPolicyOnChange) + ifNotPresent := buildModelInitCommand(true, false, true, RefreshPolicyIfNotPresent) + onChange := buildModelInitCommand(true, false, true, RefreshPolicyOnChange) Expect(onChange).To(Equal(ifNotPresent)) Expect(onChange).NotTo(ContainSubstring("--etag-compare")) }) It("does not contain user-controlled values in the OnChange command string", func() { - cmd := buildModelInitCommand(false, true, RefreshPolicyOnChange) + cmd := buildModelInitCommand(false, false, true, RefreshPolicyOnChange) Expect(cmd).NotTo(ContainSubstring("evil.com")) Expect(cmd).NotTo(ContainSubstring("touch")) }) diff --git a/internal/controller/model_storage.go b/internal/controller/model_storage.go index 927be80b..c59e39e7 100644 --- a/internal/controller/model_storage.go +++ b/internal/controller/model_storage.go @@ -163,11 +163,14 @@ func addCACertVolume(volumes *[]corev1.Volume, mounts *[]corev1.VolumeMount, cmd *cmd = fmt.Sprintf("export CURL_CA_BUNDLE=/custom-certs/$(ls /custom-certs | grep -v '^\\.' | head -n 1) && %s", *cmd) } -func buildModelInitCommand(isLocal, useCache bool, refreshPolicy string) string { +func buildModelInitCommand(isLocal, isS3, useCache bool, refreshPolicy string) string { if useCache { if isLocal { return `mkdir -p "$CACHE_DIR" && if [ ! -f "$MODEL_PATH" ]; then echo 'Copying model from local source...'; cp /host-model/model.gguf "$MODEL_PATH" && echo 'Model copied successfully'; else echo 'Model already cached, skipping copy'; fi` } + if isS3 { + return `mkdir -p "$CACHE_DIR" && if [ ! -f "$MODEL_PATH" ]; then echo 'Downloading model from S3...'; curl --aws-sigv4 "aws:amz:${AWS_REGION}:s3" -u "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" -f -L -o "$MODEL_PATH" "${AWS_ENDPOINT_URL}/${S3_BUCKET}/${S3_KEY}" && echo 'Model downloaded successfully'; else echo 'Model already cached, skipping download'; fi` + } if refreshPolicy == RefreshPolicyOnChange { return "mkdir -p \"$CACHE_DIR\" && " + remoteRevalidateScript } @@ -177,6 +180,9 @@ func buildModelInitCommand(isLocal, useCache bool, refreshPolicy string) string if isLocal { return `echo 'ERROR: Local model source requires model cache to be configured.'; exit 1` } + if isS3 { + return `if [ ! -f "$MODEL_PATH" ]; then echo 'Downloading model from S3...'; curl --aws-sigv4 "aws:amz:${AWS_REGION}:s3" -u "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" -f -L -o "$MODEL_PATH" "${AWS_ENDPOINT_URL}/${S3_BUCKET}/${S3_KEY}" && echo 'Model downloaded successfully'; else echo 'Model already exists, skipping download'; fi` + } if refreshPolicy == RefreshPolicyOnChange { return remoteRevalidateScript } @@ -210,11 +216,35 @@ const remoteRevalidateScript = `ETAG_MARKER="$(dirname "$MODEL_PATH")/.$(basenam `fi` func modelInitEnvVars(source, cacheDir, modelPath string) []corev1.EnvVar { - return []corev1.EnvVar{ + envs := []corev1.EnvVar{ {Name: "MODEL_SOURCE", Value: source}, {Name: "CACHE_DIR", Value: cacheDir}, {Name: "MODEL_PATH", Value: modelPath}, } + if isS3Source(source) { + bucket, key, err := parseS3Source(source) + if err == nil { + envs = append(envs, corev1.EnvVar{Name: "S3_BUCKET", Value: bucket}, corev1.EnvVar{Name: "S3_KEY", Value: key}) + } + } + return envs +} + +// modelEnvFrom returns EnvFrom entries for the model-downloader init container. +// When the model has a SourceSecretRef, it pulls AWS_* env vars (credentials, +// endpoint, region) from the referenced Secret. Returns nil when no secret ref +// is configured. +func modelEnvFrom(model *inferencev1alpha1.Model) []corev1.EnvFromSource { + if model.Spec.SourceSecretRef == nil { + return nil + } + return []corev1.EnvFromSource{ + { + SecretRef: &corev1.SecretEnvSource{ + LocalObjectReference: *model.Spec.SourceSecretRef, + }, + }, + } } // resolveHFSourceURL converts hf://repo-id sources to their huggingface.co @@ -589,7 +619,7 @@ func buildCachedStorageConfig(model *inferencev1alpha1.Model, isvc *inferencev1a }) } - cmd := buildModelInitCommand(isLocalModelSource(model.Spec.Source), true, model.Spec.RefreshPolicy) + cmd := buildModelInitCommand(isLocalModelSource(model.Spec.Source), isS3Source(model.Spec.Source), true, model.Spec.RefreshPolicy) env := modelInitEnvVars(model.Spec.Source, cacheDir, modelPath) addCACertVolume(&volumes, &initVolumeMounts, &cmd, caCertConfigMap) @@ -600,6 +630,7 @@ func buildCachedStorageConfig(model *inferencev1alpha1.Model, isvc *inferencev1a Image: initContainerImage, Command: []string{"sh", "-c", cmd}, Env: env, + EnvFrom: modelEnvFrom(model), VolumeMounts: initVolumeMounts, SecurityContext: initContainerSecurityContext(isvc), }, @@ -669,7 +700,7 @@ func buildEmptyDirStorageConfig(model *inferencev1alpha1.Model, isvc *inferencev }, } - cmd := buildModelInitCommand(isLocalModelSource(model.Spec.Source), false, model.Spec.RefreshPolicy) + cmd := buildModelInitCommand(isLocalModelSource(model.Spec.Source), isS3Source(model.Spec.Source), false, model.Spec.RefreshPolicy) env := modelInitEnvVars(model.Spec.Source, "", modelPath) addCACertVolume(&volumes, &initVolumeMounts, &cmd, caCertConfigMap) @@ -681,6 +712,7 @@ func buildEmptyDirStorageConfig(model *inferencev1alpha1.Model, isvc *inferencev Image: initContainerImage, Command: []string{"sh", "-c", cmd}, Env: env, + EnvFrom: modelEnvFrom(model), VolumeMounts: initVolumeMounts, SecurityContext: initContainerSecurityContext(isvc), }, diff --git a/internal/controller/model_storage_test.go b/internal/controller/model_storage_test.go new file mode 100644 index 00000000..07d04ef8 --- /dev/null +++ b/internal/controller/model_storage_test.go @@ -0,0 +1,96 @@ +/* +Copyright 2025. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + + inferencev1alpha1 "github.com/defilantech/llmkube/api/v1alpha1" +) + +var _ = Describe("buildModelInitCommand (s3)", func() { + It("should emit the --aws-sigv4 curl line for s3 source with cache", func() { + cmd := buildModelInitCommand(false, true, true, "") + Expect(cmd).To(ContainSubstring("curl --aws-sigv4")) + Expect(cmd).To(ContainSubstring("${AWS_ENDPOINT_URL}/${S3_BUCKET}/${S3_KEY}")) + Expect(cmd).To(ContainSubstring("Downloading model from S3")) + Expect(cmd).To(ContainSubstring("Model downloaded successfully")) + Expect(cmd).To(ContainSubstring("Model already cached, skipping download")) + }) + + It("should emit the --aws-sigv4 curl line for s3 source without cache", func() { + cmd := buildModelInitCommand(false, true, false, "") + Expect(cmd).To(ContainSubstring("curl --aws-sigv4")) + Expect(cmd).To(ContainSubstring("${AWS_ENDPOINT_URL}/${S3_BUCKET}/${S3_KEY}")) + Expect(cmd).To(ContainSubstring("Downloading model from S3")) + Expect(cmd).To(ContainSubstring("Model downloaded successfully")) + Expect(cmd).To(ContainSubstring("Model already exists, skipping download")) + }) + + It("should NOT emit --aws-sigv4 for non-s3 source", func() { + cmd := buildModelInitCommand(false, false, true, "") + Expect(cmd).ToNot(ContainSubstring("aws-sigv4")) + Expect(cmd).To(ContainSubstring("curl -f -L -o \"$MODEL_PATH\" \"$MODEL_SOURCE\"")) + }) + + It("should emit the --aws-sigv4 curl line for s3 source with OnChange refresh", func() { + cmd := buildModelInitCommand(false, true, true, RefreshPolicyOnChange) + Expect(cmd).To(ContainSubstring("curl --aws-sigv4")) + Expect(cmd).To(ContainSubstring("${AWS_ENDPOINT_URL}/${S3_BUCKET}/${S3_KEY}")) + }) +}) + +var _ = Describe("modelInitEnvVars (s3)", func() { + It("should include S3_BUCKET and S3_KEY for s3 source", func() { + envs := modelInitEnvVars("s3://my-bucket/models/model.gguf", "/models/cache", "/models/cache/model.gguf") + Expect(envs).To(HaveLen(5)) + Expect(envs).To(ContainElement(corev1.EnvVar{Name: "S3_BUCKET", Value: "my-bucket"})) + Expect(envs).To(ContainElement(corev1.EnvVar{Name: "S3_KEY", Value: "models/model.gguf"})) + Expect(envs).To(ContainElement(corev1.EnvVar{Name: "MODEL_SOURCE", Value: "s3://my-bucket/models/model.gguf"})) + Expect(envs).To(ContainElement(corev1.EnvVar{Name: "CACHE_DIR", Value: "/models/cache"})) + Expect(envs).To(ContainElement(corev1.EnvVar{Name: "MODEL_PATH", Value: "/models/cache/model.gguf"})) + }) + + It("should NOT include S3_BUCKET and S3_KEY for non-s3 source", func() { + envs := modelInitEnvVars("https://example.com/model.gguf", "/models/cache", "/models/cache/model.gguf") + Expect(envs).To(HaveLen(3)) + Expect(envs).ToNot(ContainElement(corev1.EnvVar{Name: "S3_BUCKET"})) + Expect(envs).ToNot(ContainElement(corev1.EnvVar{Name: "S3_KEY"})) + }) +}) + +var _ = Describe("modelEnvFrom", func() { + It("should return nil when SourceSecretRef is nil", func() { + model := &inferencev1alpha1.Model{} + envFrom := modelEnvFrom(model) + Expect(envFrom).To(BeNil()) + }) + + It("should return EnvFrom with SecretRef when SourceSecretRef is set", func() { + model := &inferencev1alpha1.Model{ + Spec: inferencev1alpha1.ModelSpec{ + SourceSecretRef: &corev1.LocalObjectReference{Name: "s3-credentials"}, + }, + } + envFrom := modelEnvFrom(model) + Expect(envFrom).To(HaveLen(1)) + Expect(envFrom[0].SecretRef).ToNot(BeNil()) + Expect(envFrom[0].SecretRef.Name).To(Equal("s3-credentials")) + }) +}) diff --git a/internal/controller/source.go b/internal/controller/source.go index d6b9998f..c627f014 100644 --- a/internal/controller/source.go +++ b/internal/controller/source.go @@ -58,6 +58,46 @@ func isPVCSource(source string) bool { return strings.HasPrefix(source, "pvc://") } +// isS3Source reports whether source is an s3:// URL. Case-folded to agree +// with the other scheme classifiers (GHSA-jw3m-8q7m-f35r). +func isS3Source(source string) bool { + return hasSchemeFold(source, "s3://") +} + +// parseS3Source splits s3://bucket/key into bucket and key. Endpoint, +// region, and credentials are NOT in the URL; they come from the +// sourceSecretRef env (AWS_ENDPOINT_URL, AWS_REGION, AWS_ACCESS_KEY_ID, +// AWS_SECRET_ACCESS_KEY). Mirrors parsePVCSource error handling. +func parseS3Source(source string) (bucket, key string, err error) { + if !isS3Source(source) { + return "", "", fmt.Errorf("not an S3 source: %s", source) + } + + // Strip the s3:// prefix + rest := strings.TrimPrefix(source, "s3://") + if rest == "" { + return "", "", fmt.Errorf("empty S3 source: %s", source) + } + + // Split into bucket and key + slashIdx := strings.Index(rest, "/") + if slashIdx < 0 { + return "", "", fmt.Errorf("S3 source must include a key: %s (expected s3://bucket/key)", source) + } + + bucket = rest[:slashIdx] + key = rest[slashIdx+1:] + + if bucket == "" { + return "", "", fmt.Errorf("S3 source has empty bucket: %s", source) + } + if key == "" { + return "", "", fmt.Errorf("S3 source has empty key: %s", source) + } + + return bucket, key, nil +} + // parsePVCSource extracts the PVC claim name and file path from a pvc:// source. // Format: pvc://claim-name/path/to/model.gguf func parsePVCSource(source string) (claimName, path string, err error) { @@ -223,6 +263,9 @@ func isHFRepoSource(source string) bool { if isPVCSource(source) { return false } + if isS3Source(source) { + return false + } if isLocalSource(source) { return false } diff --git a/internal/controller/source_test.go b/internal/controller/source_test.go index 7431f81c..5f21ad54 100644 --- a/internal/controller/source_test.go +++ b/internal/controller/source_test.go @@ -325,3 +325,79 @@ var _ = Describe("isUnrecoverableFetchError (source.go)", func() { Expect(isUnrecoverableFetchError(outer)).To(BeTrue()) }) }) + +var _ = Describe("isS3Source (source.go)", func() { + It("should return true for s3:// prefix", func() { + Expect(isS3Source("s3://my-bucket/model.gguf")).To(BeTrue()) + }) + It("should return true for case-variant S3:// prefix", func() { + Expect(isS3Source("S3://my-bucket/model.gguf")).To(BeTrue()) + }) + It("should return true for nested key", func() { + Expect(isS3Source("s3://bucket/deep/path/model.gguf")).To(BeTrue()) + }) + It("should return false for http://", func() { + Expect(isS3Source("http://example.com/model.gguf")).To(BeFalse()) + }) + It("should return false for https://", func() { + Expect(isS3Source("https://example.com/model.gguf")).To(BeFalse()) + }) + It("should return false for empty string", func() { + Expect(isS3Source("")).To(BeFalse()) + }) +}) + +var _ = Describe("parseS3Source (source.go)", func() { + It("should parse simple s3 source", func() { + bucket, key, err := parseS3Source("s3://my-bucket/model.gguf") + Expect(err).NotTo(HaveOccurred()) + Expect(bucket).To(Equal("my-bucket")) + Expect(key).To(Equal("model.gguf")) + }) + + It("should parse nested key", func() { + bucket, key, err := parseS3Source("s3://shared-bucket/models/llama/7b/model.gguf") + Expect(err).NotTo(HaveOccurred()) + Expect(bucket).To(Equal("shared-bucket")) + Expect(key).To(Equal("models/llama/7b/model.gguf")) + }) + + It("should error on non-S3 source", func() { + _, _, err := parseS3Source("http://example.com/model.gguf") + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("not an S3 source")) + }) + + It("should error on empty S3 source", func() { + _, _, err := parseS3Source("s3://") + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("empty S3 source")) + }) + + It("should error on missing key", func() { + _, _, err := parseS3Source("s3://my-bucket") + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("must include a key")) + }) + + It("should error on empty bucket", func() { + _, _, err := parseS3Source("s3:///model.gguf") + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("empty bucket")) + }) + + It("should error on trailing slash only (empty key)", func() { + _, _, err := parseS3Source("s3://my-bucket/") + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("empty key")) + }) +}) + +var _ = Describe("isHFRepoSource rejects s3:// (source.go)", func() { + It("should return false for s3:// source", func() { + Expect(isHFRepoSource("s3://my-bucket/model.gguf")).To(BeFalse()) + }) + It("should return false for case-variant S3:// source", func() { + Expect(isHFRepoSource("S3://my-bucket/model.gguf")).To(BeFalse()) + }) +})