Merge pull request #136 from sgotti/implement_container_volume_tmpfs

*: implement ability to add tmpfs volumes to containers

internal/config/config.go

@@ -26,6 +26,7 @@ import (
 	"github.com/ghodss/yaml"
 	"github.com/google/go-jsonnet"
 	errors "golang.org/x/xerrors"
+	"k8s.io/apimachinery/pkg/api/resource"
 )
 
 const (
@@ -92,6 +93,17 @@ type Container struct {
 	User        string           `json:"user"`
 	Privileged  bool             `json:"privileged"`
 	Entrypoint  string           `json:"entrypoint"`
+	Volumes     []Volume         `json:"volumes"`
+}
+
+type Volume struct {
+	Path string `json:"path"`
+
+	TmpFS *VolumeTmpFS `json:"tmpfs"`
+}
+
+type VolumeTmpFS struct {
+	Size *resource.Quantity `json:"size"`
 }
 
 type Run struct {
@@ -711,6 +723,14 @@ func checkConfig(config *Config) error {
 					return errors.Errorf("task %q runtime: invalid arch %q", task.Name, r.Arch)
 				}
 			}
+
+			for _, container := range r.Containers {
+				for _, vol := range container.Volumes {
+					if vol.TmpFS == nil {
+						return errors.Errorf("no volume config specified")
+					}
+				}
+			}
 		}
 	}
 

internal/config/config_test.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	errors "golang.org/x/xerrors"
+	"k8s.io/apimachinery/pkg/api/resource"
 )
 
 func TestParseConfig(t *testing.T) {
@@ -324,11 +325,18 @@ func TestParseOutput(t *testing.T) {
                           type: pod
                           containers:
                             - image: image01
+                              volumes:
+                                - path: /mnt/tmpfs
+                                  tmpfs:
+                                    size: 1Gi
                       - name: task04
                         runtime:
                           type: pod
                           containers:
                             - image: image01
+                              volumes:
+                                - path: /mnt/tmpfs
+                                  tmpfs: {}
           `,
 			out: &Config{
 				Runs: []*Run{
@@ -489,6 +497,7 @@ func TestParseOutput(t *testing.T) {
 									Containers: []*Container{
 										&Container{
 											Image:   "image01",
+											Volumes: []Volume{{Path: "/mnt/tmpfs", TmpFS: &VolumeTmpFS{Size: resource.NewQuantity(1024*1024*1024, resource.BinarySI)}}},
 										},
 									},
 								},
@@ -504,6 +513,7 @@ func TestParseOutput(t *testing.T) {
 									Containers: []*Container{
 										&Container{
 											Image:   "image01",
+											Volumes: []Volume{{Path: "/mnt/tmpfs", TmpFS: &VolumeTmpFS{}}},
 										},
 									},
 								},
@@ -524,7 +534,16 @@ func TestParseOutput(t *testing.T) {
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
-			if diff := cmp.Diff(tt.out, out); diff != "" {
+			if diff := cmp.Diff(tt.out, out, cmp.Comparer(func(x, y *resource.Quantity) bool {
+				if x == nil && y == nil {
+					return true
+				}
+				if x != nil && y != nil {
+					return x.Cmp(*y) == 0
+				}
+
+				return false
+			})); diff != "" {
 				t.Error(diff)
 			}
 		})

internal/runconfig/runconfig.go

@@ -40,8 +40,24 @@ func genRuntime(c *config.Config, ce *config.Runtime, variables map[string]strin
 			User:        cc.User,
 			Privileged:  cc.Privileged,
 			Entrypoint:  cc.Entrypoint,
+			Volumes:     make([]rstypes.Volume, len(cc.Volumes)),
 		}
 
+		for i, ccVol := range cc.Volumes {
+			container.Volumes[i] = rstypes.Volume{
+				Path: ccVol.Path,
+			}
+
+			if ccVol.TmpFS != nil {
+				var size int64
+				if ccVol.TmpFS.Size != nil {
+					size = ccVol.TmpFS.Size.Value()
+				}
+				container.Volumes[i].TmpFS = &rstypes.VolumeTmpFS{
+					Size: size,
+				}
+			}
+		}
 		containers = append(containers, container)
 	}
 

internal/runconfig/runconfig_test.go

@@ -23,6 +23,7 @@ import (
 	"agola.io/agola/internal/util"
 	rstypes "agola.io/agola/services/runservice/types"
 	"agola.io/agola/services/types"
+	"k8s.io/apimachinery/pkg/api/resource"
 
 	"github.com/google/go-cmp/cmp"
 	errors "golang.org/x/xerrors"
@@ -721,6 +722,16 @@ func TestGenRunConfig(t *testing.T) {
 												"ENVFROMVARIABLE01": config.Value{Type: config.ValueTypeFromVariable, Value: "variable01"},
 											},
 											User: "",
+											Volumes: []config.Volume{
+												config.Volume{
+													Path:  "/mnt/vol01",
+													TmpFS: &config.VolumeTmpFS{},
+												},
+												config.Volume{
+													Path:  "/mnt/vol01",
+													TmpFS: &config.VolumeTmpFS{Size: resource.NewQuantity(1024*1024*1024, resource.BinarySI)},
+												},
+											},
 										},
 									},
 								},
@@ -798,6 +809,16 @@ func TestGenRunConfig(t *testing.T) {
 									"ENV01":             "ENV01",
 									"ENVFROMVARIABLE01": "VARVALUE01",
 								},
+								Volumes: []rstypes.Volume{
+									rstypes.Volume{
+										Path:  "/mnt/vol01",
+										TmpFS: &rstypes.VolumeTmpFS{},
+									},
+									rstypes.Volume{
+										Path:  "/mnt/vol01",
+										TmpFS: &rstypes.VolumeTmpFS{Size: 1024 * 1024 * 1024},
+									},
+								},
 							},
 						},
 					},
@@ -874,6 +895,7 @@ func TestGenRunConfig(t *testing.T) {
 							{
 								Image:       "image01",
 								Environment: map[string]string{},
+								Volumes:     []rstypes.Volume{},
 							},
 						},
 					},
@@ -972,6 +994,7 @@ func TestGenRunConfig(t *testing.T) {
 							{
 								Image:       "image01",
 								Environment: map[string]string{},
+								Volumes:     []rstypes.Volume{},
 							},
 						},
 					},
@@ -989,9 +1012,6 @@ func TestGenRunConfig(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			out := GenRunConfigTasks(uuid, tt.in, "run01", tt.variables, "", "", "")
 
-			//if err != nil {
-			//	t.Fatalf("unexpected error: %v", err)
-			//}
 			if diff := cmp.Diff(tt.out, out); diff != "" {
 				t.Error(diff)
 			}

internal/services/executor/driver/docker.go

@@ -36,6 +36,7 @@ import (
 	dockertypes "github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
+	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/client"
 	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/pkg/stdcopy"
@@ -285,6 +286,7 @@ func (d *DockerDriver) createContainer(ctx context.Context, index int, podConfig
 	}
 	if index == 0 {
 		// main container requires the initvolume containing the toolbox
+		// TODO(sgotti) migrate this to cliHostConfig.Mounts
 		cliHostConfig.Binds = []string{fmt.Sprintf("%s:%s", d.initVolumeHostDir, podConfig.InitVolumeDir)}
 		cliHostConfig.ReadonlyPaths = []string{fmt.Sprintf("%s:%s", d.initVolumeHostDir, podConfig.InitVolumeDir)}
 	} else {
@@ -292,6 +294,22 @@ func (d *DockerDriver) createContainer(ctx context.Context, index int, podConfig
 		cliHostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", maincontainerID))
 	}
 
+	for _, vol := range containerConfig.Volumes {
+		if vol.TmpFS != nil {
+			cliHostConfig.Mounts = []mount.Mount{
+				mount.Mount{
+					Type:   mount.TypeTmpfs,
+					Target: vol.Path,
+					TmpfsOptions: &mount.TmpfsOptions{
+						SizeBytes: vol.TmpFS.Size,
+					},
+				},
+			}
+		} else {
+			return nil, errors.Errorf("missing volume config")
+		}
+	}
+
 	resp, err := d.client.ContainerCreate(ctx, cliContainerConfig, cliHostConfig, nil, "")
 	return &resp, err
 }

internal/services/executor/driver/docker_test.go

@@ -370,4 +370,84 @@ func TestDockerPod(t *testing.T) {
 			t.Fatalf("pod with id %q not found", pod.ID())
 		}
 	})
+
+	t.Run("test pod with a tmpfs volume with size limit", func(t *testing.T) {
+		pod, err := d.NewPod(ctx, &PodConfig{
+			ID:     uuid.NewV4().String(),
+			TaskID: uuid.NewV4().String(),
+			Containers: []*ContainerConfig{
+				&ContainerConfig{
+					Cmd:   []string{"cat"},
+					Image: "busybox",
+					Volumes: []Volume{
+						{
+							Path: "/mnt/tmpfs",
+							TmpFS: &VolumeTmpFS{
+								Size: 1024 * 1024,
+							},
+						},
+					},
+				},
+			},
+			InitVolumeDir: "/tmp/agola",
+		}, ioutil.Discard)
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		defer func() { _ = pod.Remove(ctx) }()
+
+		ce, err := pod.Exec(ctx, &ExecConfig{
+			Cmd: []string{"sh", "-c", "if [ $(cat /proc/mounts | grep /mnt/tmpfs | grep size=1024k | wc -l ) -ne 1 ]; then exit 1; fi"},
+		})
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+
+		code, err := ce.Wait(ctx)
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		if code != 0 {
+			t.Fatalf("unexpected exit code: %d", code)
+		}
+	})
+
+	t.Run("test pod with a tmpfs volume without size limit", func(t *testing.T) {
+		pod, err := d.NewPod(ctx, &PodConfig{
+			ID:     uuid.NewV4().String(),
+			TaskID: uuid.NewV4().String(),
+			Containers: []*ContainerConfig{
+				&ContainerConfig{
+					Cmd:   []string{"cat"},
+					Image: "busybox",
+					Volumes: []Volume{
+						{
+							Path:  "/mnt/tmpfs",
+							TmpFS: &VolumeTmpFS{},
+						},
+					},
+				},
+			},
+			InitVolumeDir: "/tmp/agola",
+		}, ioutil.Discard)
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		defer func() { _ = pod.Remove(ctx) }()
+
+		ce, err := pod.Exec(ctx, &ExecConfig{
+			Cmd: []string{"sh", "-c", "if [ $(cat /proc/mounts | grep /mnt/tmpfs | wc -l ) -ne 1 ]; then exit 1; fi"},
+		})
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+
+		code, err := ce.Wait(ctx)
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		if code != 0 {
+			t.Fatalf("unexpected exit code: %d", code)
+		}
+	})
 }

internal/services/executor/driver/driver.go

@@ -93,6 +93,17 @@ type ContainerConfig struct {
 	Image      string
 	User       string
 	Privileged bool
+	Volumes    []Volume
+}
+
+type Volume struct {
+	Path string
+
+	TmpFS *VolumeTmpFS
+}
+
+type VolumeTmpFS struct {
+	Size int64
 }
 
 type ExecConfig struct {

internal/services/executor/driver/k8s.go

@@ -35,6 +35,7 @@ import (
 	errors "golang.org/x/xerrors"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	apilabels "k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/watch"
@@ -416,6 +417,37 @@ func (d *K8sDriver) NewPod(ctx context.Context, podConfig *PodConfig, out io.Wri
 				},
 			}
 		}
+
+		for vIndex, cVol := range containerConfig.Volumes {
+			var vol corev1.Volume
+			var volMount corev1.VolumeMount
+			if cVol.TmpFS != nil {
+				name := fmt.Sprintf("volume-%d-%d", cIndex, vIndex)
+				var sizeLimit *resource.Quantity
+				if cVol.TmpFS.Size != 0 {
+					sizeLimit = resource.NewQuantity(cVol.TmpFS.Size, resource.BinarySI)
+				}
+				vol = corev1.Volume{
+					Name: name,
+					VolumeSource: corev1.VolumeSource{
+						EmptyDir: &corev1.EmptyDirVolumeSource{
+							Medium:    corev1.StorageMediumMemory,
+							SizeLimit: sizeLimit,
+						},
+					},
+				}
+				volMount = corev1.VolumeMount{
+					Name:      name,
+					MountPath: cVol.Path,
+				}
+			} else {
+				return nil, errors.Errorf("missing volume config")
+			}
+
+			pod.Spec.Volumes = append(pod.Spec.Volumes, vol)
+			c.VolumeMounts = append(c.VolumeMounts, volMount)
+		}
+
 		pod.Spec.Containers = append(pod.Spec.Containers, c)
 	}
 

internal/services/executor/driver/k8s_test.go

@@ -259,6 +259,49 @@ func TestK8sPod(t *testing.T) {
 		}
 	})
 
+	t.Run("test pod with a tmpfs volume", func(t *testing.T) {
+		pod, err := d.NewPod(ctx, &PodConfig{
+			ID:     uuid.NewV4().String(),
+			TaskID: uuid.NewV4().String(),
+			Containers: []*ContainerConfig{
+				&ContainerConfig{
+					Cmd:   []string{"cat"},
+					Image: "busybox",
+					Volumes: []Volume{
+						{
+							Path: "/mnt/tmpfs",
+							TmpFS: &VolumeTmpFS{
+								Size: 1024 * 1024,
+							},
+						},
+					},
+				},
+			},
+			InitVolumeDir: "/tmp/agola",
+		}, ioutil.Discard)
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		defer func() { _ = pod.Remove(ctx) }()
+
+		var buf bytes.Buffer
+		ce, err := pod.Exec(ctx, &ExecConfig{
+			// k8s doesn't set size=1024k in the tmpf mount options but uses other modes to detect the size
+			Cmd:    []string{"sh", "-c", "if [ $(cat /proc/mounts | grep /mnt/tmpfs | wc -l ) -ne 1 ]; then exit 1; fi"},
+			Stdout: &buf,
+		})
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+
+		code, err := ce.Wait(ctx)
+		if err != nil {
+			t.Fatalf("unexpected err: %v", err)
+		}
+		if code != 0 {
+			t.Fatalf("unexpected exit code: %d", code)
+		}
+	})
 }
 
 func TestParseGitVersion(t *testing.T) {

internal/services/executor/executor.go

@@ -893,13 +893,27 @@ func (e *Executor) setupTask(ctx context.Context, rt *runningTask) error {
 			cmd = strings.Split(c.Entrypoint, " ")
 		}
 
-		podConfig.Containers[i] = &driver.ContainerConfig{
+		containerConfig := &driver.ContainerConfig{
 			Image:      c.Image,
 			Cmd:        cmd,
 			Env:        c.Environment,
 			User:       c.User,
 			Privileged: c.Privileged,
+			Volumes:    make([]driver.Volume, len(c.Volumes)),
 		}
+
+		for vIndex, cVol := range c.Volumes {
+			containerConfig.Volumes[vIndex] = driver.Volume{
+				Path: cVol.Path,
+			}
+			if cVol.TmpFS != nil {
+				containerConfig.Volumes[vIndex].TmpFS = &driver.VolumeTmpFS{
+					Size: cVol.TmpFS.Size,
+				}
+			}
+		}
+
+		podConfig.Containers[i] = containerConfig
 	}
 
 	_, _ = outf.WriteString("Starting pod.\n")

services/runservice/types/types.go

@@ -543,6 +543,17 @@ type Container struct {
 	User        string            `json:"user,omitempty"`
 	Privileged  bool              `json:"privileged"`
 	Entrypoint  string            `json:"entrypoint"`
+	Volumes     []Volume          `json:"volumes"`
+}
+
+type Volume struct {
+	Path string `json:"path"`
+
+	TmpFS *VolumeTmpFS `json:"tmpfs"`
+}
+
+type VolumeTmpFS struct {
+	Size int64 `json:"size"`
 }
 
 type WorkspaceOperation struct {