diff --git a/README.md b/README.md index eea87aa..f604e6b 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,8 @@ Runtime selection is daemon-only: start the daemon with `druid daemon --runtime Kubernetes runtime support is available with `druid daemon --runtime kubernetes` for in-cluster daemons or out-of-cluster daemons using kubeconfig. It stores daemon scroll state in ConfigMaps, materializes OCI artifacts through `druid worker pull` Jobs, and uses kubelet pod stats for procedure-level traffic checks. See `docs/kubernetes_runtime.md` for kubeconfig, RBAC, and PVC setup. +Both backends refresh registry-tagged container images at startup, including `stable`, `latest`, and version tags. Kubernetes uses `imagePullPolicy: Always`; Docker pulls before creating the container. Existing layers are reused when the registry digest is unchanged. Digest-pinned images and explicit `:local` development images use the cache when available. A tag update takes effect on the next container start; running containers are not automatically restarted. + ## Documentation Read more at https://docs.druid.gg/cli diff --git a/internal/runtime/docker/images.go b/internal/runtime/docker/images.go index 87bb6e0..7f3b779 100644 --- a/internal/runtime/docker/images.go +++ b/internal/runtime/docker/images.go @@ -2,23 +2,40 @@ package docker import ( "context" + "encoding/json" + "fmt" "io" cerrdefs "github.com/containerd/errdefs" "github.com/docker/docker/api/types/image" + "github.com/highcard-dev/daemon/internal/runtime/imagepolicy" ) func (b *Backend) pullImage(ctx context.Context, imageRef string) error { - if _, err := b.client.ImageInspect(ctx, imageRef); err == nil { - return nil - } else if !cerrdefs.IsNotFound(err) { - return err + if !imagepolicy.ShouldRefresh(imageRef) { + if _, err := b.client.ImageInspect(ctx, imageRef); err == nil { + return nil + } else if !cerrdefs.IsNotFound(err) { + return err + } } reader, err := b.client.ImagePull(ctx, imageRef, image.PullOptions{}) if err != nil { return err } defer reader.Close() - _, _ = io.Copy(io.Discard, reader) - return nil + decoder := json.NewDecoder(reader) + for { + var message struct { + Error string `json:"error"` + } + if err := decoder.Decode(&message); err == io.EOF { + return nil + } else if err != nil { + return err + } + if message.Error != "" { + return fmt.Errorf("pull image %s: %s", imageRef, message.Error) + } + } } diff --git a/internal/runtime/docker/images_test.go b/internal/runtime/docker/images_test.go new file mode 100644 index 0000000..699c236 --- /dev/null +++ b/internal/runtime/docker/images_test.go @@ -0,0 +1,81 @@ +package docker + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/docker/docker/client" +) + +func TestPullImageRefreshesRegistryTags(t *testing.T) { + digestImage := "druid@sha256:" + strings.Repeat("a", 64) + for _, tc := range []struct { + name string + image string + cached bool + pullBody string + pullStatus int + wantPulls int + wantError string + }{ + {name: "stable refreshes cached image", image: "registry.example/druid:stable", cached: true, wantPulls: 1}, + {name: "version tags also refresh", image: "alpine:3.20", cached: true, wantPulls: 1}, + {name: "digest reuses cache", image: digestImage, cached: true}, + {name: "local image reuses cache", image: "druid:local", cached: true}, + {name: "missing digest pulls", image: digestImage, wantPulls: 1}, + {name: "registry failure does not use stale cache", image: "druid:stable", cached: true, pullStatus: http.StatusInternalServerError, pullBody: `{"message":"registry unavailable"}`, wantPulls: 1, wantError: "registry unavailable"}, + {name: "stream failure does not use stale cache", image: "druid:stable", cached: true, pullBody: `{"error":"access denied"}`, wantPulls: 1, wantError: "access denied"}, + } { + t.Run(tc.name, func(t *testing.T) { + pulls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/json"): + if !tc.cached { + w.WriteHeader(http.StatusNotFound) + fmt.Fprint(w, `{"message":"image not found"}`) + return + } + fmt.Fprint(w, `{"Id":"sha256:cached"}`) + case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/images/create"): + pulls++ + status := tc.pullStatus + if status == 0 { + status = http.StatusOK + } + w.WriteHeader(status) + if tc.pullBody != "" { + fmt.Fprint(w, tc.pullBody) + } else { + fmt.Fprint(w, `{"status":"downloaded current image"}`) + } + default: + t.Errorf("unexpected Docker request: %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + cli, err := client.NewClientWithOpts(client.WithHost(server.URL), client.WithVersion("1.47")) + if err != nil { + t.Fatal(err) + } + defer cli.Close() + backend := &Backend{client: cli} + err = backend.pullImage(context.Background(), tc.image) + if tc.wantError == "" && err != nil { + t.Fatal(err) + } + if tc.wantError != "" && (err == nil || !strings.Contains(err.Error(), tc.wantError)) { + t.Fatalf("error = %v, want %q", err, tc.wantError) + } + if pulls != tc.wantPulls { + t.Fatalf("pull requests = %d, want %d", pulls, tc.wantPulls) + } + }) + } +} diff --git a/internal/runtime/imagepolicy/policy.go b/internal/runtime/imagepolicy/policy.go new file mode 100644 index 0000000..4f148f5 --- /dev/null +++ b/internal/runtime/imagepolicy/policy.go @@ -0,0 +1,9 @@ +package imagepolicy + +import "strings" + +// ShouldRefresh checks registry tags on each container start. Only digests are +// immutable; :local images are explicitly built and imported for development. +func ShouldRefresh(image string) bool { + return !strings.Contains(image, "@") && !strings.HasSuffix(image, ":local") +} diff --git a/internal/runtime/imagepolicy/policy_test.go b/internal/runtime/imagepolicy/policy_test.go new file mode 100644 index 0000000..69ce902 --- /dev/null +++ b/internal/runtime/imagepolicy/policy_test.go @@ -0,0 +1,27 @@ +package imagepolicy + +import "testing" + +func TestShouldRefresh(t *testing.T) { + for _, tc := range []struct { + image string + want bool + }{ + {"artifacts.druid.gg/druid-team/druid-dev:stable", true}, + {"druid:latest", true}, + {"alpine", true}, + {"alpine:3.20", true}, + {"registry.example:5000/team/image:release", true}, + {"registry.example:5000/team/image", true}, + {"druid@sha256:abc", false}, + {"druid:stable@sha256:abc", false}, + {"druid:local", false}, + {"registry.example:5000/druid:local", false}, + } { + t.Run(tc.image, func(t *testing.T) { + if got := ShouldRefresh(tc.image); got != tc.want { + t.Fatalf("ShouldRefresh(%q) = %v, want %v", tc.image, got, tc.want) + } + }) + } +} diff --git a/internal/runtime/kubernetes/image_pull_policy_test.go b/internal/runtime/kubernetes/image_pull_policy_test.go new file mode 100644 index 0000000..c95c8ab --- /dev/null +++ b/internal/runtime/kubernetes/image_pull_policy_test.go @@ -0,0 +1,49 @@ +package kubernetes + +import ( + "testing" + + "github.com/highcard-dev/daemon/internal/core/domain" + corev1 "k8s.io/api/core/v1" +) + +func TestRuntimeWorkloadsUseImagePullPolicy(t *testing.T) { + for _, tc := range []struct { + image string + want corev1.PullPolicy + }{ + {"artifacts.druid.gg/druid-team/druid-dev:stable", corev1.PullAlways}, + {"alpine:3.20", corev1.PullAlways}, + {"druid@sha256:abc", corev1.PullIfNotPresent}, + {"druid:local", corev1.PullIfNotPresent}, + } { + t.Run(tc.image, func(t *testing.T) { + root := ref("games", "runtime-data") + procedure := &domain.Procedure{Image: tc.image} + job, err := procedureJobSpec("games", root, "build", "build", "build-0", 0, procedure, nil, "", "") + if err != nil { + t.Fatal(err) + } + initName, mainName := "init", "files" + command := &domain.CommandInstructionSet{Procedures: []*domain.Procedure{ + {Id: &initName, Image: tc.image}, + {Id: &mainName, Image: tc.image}, + }} + sts, err := persistentStatefulSetSpec("games", root, "dev", "dev", command, nil, "") + if err != nil { + t.Fatal(err) + } + helper := helperJobSpec("games", "helper", "runtime-data", tc.image, nil, "", nil) + for _, container := range []corev1.Container{ + job.Spec.Template.Spec.Containers[0], + sts.Spec.Template.Spec.Containers[0], + sts.Spec.Template.Spec.InitContainers[0], + helper.Spec.Template.Spec.Containers[0], + } { + if container.ImagePullPolicy != tc.want { + t.Errorf("container %s (%s) pull policy = %s, want %s", container.Name, container.Image, container.ImagePullPolicy, tc.want) + } + } + }) + } +} diff --git a/internal/runtime/kubernetes/resources.go b/internal/runtime/kubernetes/resources.go index 5949af7..9f23521 100644 --- a/internal/runtime/kubernetes/resources.go +++ b/internal/runtime/kubernetes/resources.go @@ -14,8 +14,16 @@ import ( "github.com/highcard-dev/daemon/internal/core/domain" "github.com/highcard-dev/daemon/internal/core/ports" + "github.com/highcard-dev/daemon/internal/runtime/imagepolicy" ) +func imagePullPolicy(image string) corev1.PullPolicy { + if imagepolicy.ShouldRefresh(image) { + return corev1.PullAlways + } + return corev1.PullIfNotPresent +} + func pvcSpec(namespace string, name string, storageClass string, storageRequest string) *corev1.PersistentVolumeClaim { if storageRequest == "" { storageRequest = "1Gi" @@ -162,10 +170,11 @@ func helperJobSpec(namespace string, jobName string, pvc string, image string, c podSpec := corev1.PodSpec{ RestartPolicy: corev1.RestartPolicyNever, Containers: []corev1.Container{{ - Name: "main", - Image: image, - Command: command, - VolumeMounts: []corev1.VolumeMount{{Name: "scroll", MountPath: "/scroll"}}, + Name: "main", + Image: image, + ImagePullPolicy: imagePullPolicy(image), + Command: command, + VolumeMounts: []corev1.VolumeMount{{Name: "scroll", MountPath: "/scroll"}}, }}, Volumes: []corev1.Volume{pvcVolume("scroll", pvc)}, } @@ -209,7 +218,7 @@ func procedureJobSpec(namespace string, root string, commandName string, procedu WorkingDir: procedure.WorkingDir, TTY: procedure.TTY, Stdin: true, - ImagePullPolicy: corev1.PullIfNotPresent, + ImagePullPolicy: imagePullPolicy(procedure.Image), Env: envVars(env), VolumeMounts: volumeMounts(procedure.Mounts), } @@ -292,7 +301,7 @@ func persistentStatefulSetSpec(namespace string, root string, commandName string WorkingDir: mainProcedure.WorkingDir, TTY: mainProcedure.TTY, Stdin: true, - ImagePullPolicy: corev1.PullIfNotPresent, + ImagePullPolicy: imagePullPolicy(mainProcedure.Image), Env: envVars(mainEnv), VolumeMounts: volumeMounts(mainProcedure.Mounts), } @@ -316,7 +325,7 @@ func persistentStatefulSetSpec(namespace string, root string, commandName string WorkingDir: procedure.WorkingDir, TTY: procedure.TTY, Stdin: true, - ImagePullPolicy: corev1.PullIfNotPresent, + ImagePullPolicy: imagePullPolicy(procedure.Image), Env: envVars(env), VolumeMounts: volumeMounts(procedure.Mounts), }) diff --git a/test/integration/docker/docker_cli_test.go b/test/integration/docker/docker_cli_test.go index 71aeb73..05c4a3c 100644 --- a/test/integration/docker/docker_cli_test.go +++ b/test/integration/docker/docker_cli_test.go @@ -23,7 +23,7 @@ func TestDockerBackendCLIComplexLifecycle(t *testing.T) { callbackPort := e2e.FreePort(t) name := fmt.Sprintf("docker-cli-%d", time.Now().UnixNano()) fixture := e2e.WriteFixture(t, filepath.Join(t.TempDir(), "scroll"), name, port, routePort) - workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e:"+name) + workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e-"+name+":local") containerHost := e2e.DockerHostAddress(t) socket := filepath.Join(os.TempDir(), fmt.Sprintf("druid-docker-%d.sock", time.Now().UnixNano())) @@ -94,7 +94,7 @@ func TestDockerBackendVolumeStorageWorkerLifecycleBackupRestore(t *testing.T) { containerHost := e2e.DockerHostAddress(t) name := fmt.Sprintf("docker-volume-%d", time.Now().UnixNano()) fixture := e2e.WriteFixture(t, filepath.Join(t.TempDir(), "scroll"), name, port, routePort) - workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e:"+name) + workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e-"+name+":local") pushArtifact := fmt.Sprintf("127.0.0.1:%d/druid-e2e/%s:v1", registryPort, name) runtimeArtifact := fmt.Sprintf("%s:%d/druid-e2e/%s:v1", containerHost, registryPort, name) @@ -166,7 +166,7 @@ func TestDockerBackendColdstarterFrontsRuntime(t *testing.T) { publicPort := e2e.FreePort(t) callbackPort := e2e.FreePort(t) name := fmt.Sprintf("docker-coldstart-%d", time.Now().UnixNano()) - image := e2e.BuildDockerImage(t, "druid-coldstart-e2e:"+name) + image := e2e.BuildDockerImage(t, "druid-coldstart-e2e-"+name+":local") fixtureDir := writeColdstarterFixture(t, filepath.Join(t.TempDir(), "scroll"), name, image, runtimePort) containerHost := e2e.DockerHostAddress(t) diff --git a/test/integration/kubernetes/kubernetes_cli_test.go b/test/integration/kubernetes/kubernetes_cli_test.go index 182b20e..60c8d50 100644 --- a/test/integration/kubernetes/kubernetes_cli_test.go +++ b/test/integration/kubernetes/kubernetes_cli_test.go @@ -29,7 +29,7 @@ func TestKubernetesBackendCLIComplexLifecycle(t *testing.T) { namespace := "druid-cli-e2e-" + suffix name := "k8s-cli-" + suffix fixture := e2e.WriteFixture(t, filepath.Join(t.TempDir(), "scroll"), name, port, routePort) - workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e:"+name) + workerImage := e2e.BuildDockerImage(t, "druid-cli-e2e-"+name+":local") importImageIntoCurrentCluster(t, workerImage) pushArtifact := fmt.Sprintf("127.0.0.1:%d/druid-e2e/%s:v1", registryPort, name) runtimeArtifact := fmt.Sprintf("%s:%d/druid-e2e/%s:v1", containerHost, registryPort, name)