feat: Cache workspace blob on tofu-controller filesystem

charts/tf-controller/README.md

@@ -36,6 +36,7 @@ __Note__: If you need to use the `imagePullSecrets` it would be best to set `ser
 | awsPackage.install | bool | `true` |  |
 | awsPackage.repository | string | `"ghcr.io/flux-iac/aws-primitive-modules"` |  |
 | awsPackage.tag | string | `"v4.38.0-v1alpha11"` |  |
+| blobCacheVolume.emptyDir | object | `{}` |  |
 | branchPlanner | object | `{"configMap":"branch-planner","enabled":false,"image":{"pullPolicy":"IfNotPresent","repository":"ghcr.io/flux-iac/branch-planner","tag":""},"podSecurityContext":{"fsGroup":1337},"pollingInterval":"30s","resources":{"limits":{"cpu":"1000m","memory":"1Gi"},"requests":{"cpu":"200m","memory":"64Mi"}},"securityContext":{"allowPrivilegeEscalation":false,"capabilities":{"drop":["ALL"]},"readOnlyRootFilesystem":true,"runAsNonRoot":true,"runAsUser":65532,"seccompProfile":{"type":"RuntimeDefault"}},"sourceInterval":"30s"}` | Branch Planner-specific configurations |
 | caCertValidityDuration | string | `"168h0m"` | Argument for `--ca-cert-validity-duration` (Controller) |
 | certRotationCheckFrequency | string | `"30m0s"` | Argument for `--cert-rotation-check-frequency` (Controller) |

charts/tf-controller/templates/deployment.yaml

@@ -27,6 +27,9 @@ spec:
         {{- toYaml . | nindent 8 }}
       {{- end }}
       {{- end }}
+      volumes:
+      - name: blob-cache
+        {{- toYaml .Values.blobCacheVolume | nindent 8 }}
       containers:
       - args:
         - --allow-cross-namespace-refs={{ .Values.allowCrossNamespaceRefs }}
@@ -79,8 +82,10 @@ spec:
           {{- toYaml .Values.resources | nindent 10 }}
         securityContext:
           {{- toYaml .Values.securityContext | nindent 10 }}
-        {{- with .Values.volumeMounts }}
         volumeMounts:
+          - name: blob-cache
+            mountPath: /blob-cache/
+        {{- with .Values.volumeMounts }}
           {{- toYaml . | nindent 10 }}
         {{- end }}
       securityContext:

charts/tf-controller/templates/runner-serviceaccount.yaml

@@ -15,6 +15,7 @@ metadata:
 {{- with $.Values.imagePullSecrets }}
 imagePullSecrets:
   {{- toYaml . | nindent 2 }}
+{{- end }}
 ---
 apiVersion: v1
 kind: Secret
@@ -26,4 +27,3 @@ metadata:
 type: kubernetes.io/service-account-token
 {{- end }}
 {{- end }}
-{{- end }}

charts/tf-controller/values.yaml

@@ -112,6 +112,10 @@ awsPackage:
   install: true
   tag: v4.38.0-v1alpha11
   repository: ghcr.io/flux-iac/aws-primitive-modules
+blobCacheVolume:
+  emptyDir: {}
+  # persistentVolumeClaim:
+  #   claimName: tofu-controller-blob-cache
 # -- Runner-specific configurations
 runner:
   image:

controllers/filesystem.go

@@ -0,0 +1,27 @@
+package controllers
+
+import (
+	"io"
+	"os"
+	"path"
+)
+
+type Filesystem interface {
+	GetWriter(filePath string) (io.WriteCloser, error)
+	GetReader(filePath string) (io.ReadCloser, error)
+}
+
+type LocalFilesystem struct {
+	root string
+}
+
+func NewLocalFilesystem(root string) *LocalFilesystem {
+	return &LocalFilesystem{root: root}
+}
+
+func (fs *LocalFilesystem) GetWriter(filePath string) (io.WriteCloser, error) {
+	return os.Create(path.Join(fs.root, filePath))
+}
+func (fs *LocalFilesystem) GetReader(filePath string) (io.ReadCloser, error) {
+	return os.Open(path.Join(fs.root, filePath))
+}

controllers/tf_controller_backend.go

@@ -421,6 +421,11 @@ terraform {
 		}
 	}
 
+	// Save workspace after init
+	if err := r.GetWorkspaceBlobCache(ctx, runnerClient, &terraform, tfInstance, workingDir); err != nil {
+		log.Error(err, "unable to get workspace blob cache")
+	}
+
 	return terraform, tfInstance, tmpDir, nil
 }
 

controllers/tf_controller_blob_cache.go

@@ -0,0 +1,86 @@
+package controllers
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"io"
+
+	infrav1 "github.com/flux-iac/tofu-controller/api/v1alpha2"
+	"github.com/flux-iac/tofu-controller/runner"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+// TODO: This path is hard-coded in the Deployment of the controller as mount
+// point. Why not /tmp? Good question, it's a ro mount so we can't write there.
+const cacheDir = "/blob-cache"
+
+func (r *TerraformReconciler) GetWorkspaceBlobCache(ctx context.Context, runnerClient runner.RunnerClient, terraform *infrav1.Terraform, tfInstance, workdir string) error {
+	log := ctrl.LoggerFrom(ctx).WithValues("step", "get workspace blob cache")
+
+	log.Info("request workspace blob from runner", "workdir", workdir, "tfInstance", tfInstance)
+	streamClient, err := runnerClient.CreateWorkspaceBlobStream(ctx, &runner.CreateWorkspaceBlobRequest{
+		TfInstance: tfInstance,
+		WorkingDir: workdir,
+		Namespace:  terraform.Namespace,
+	})
+	if err != nil {
+		return err
+	}
+
+	fs := NewLocalFilesystem(cacheDir)
+	sha := sha256.New()
+	checksum := []byte{}
+
+	// TODO: This file pattern needs some love, it's there as a placeholder.
+	// It would be beneficial if we can add the commit hash to the filename, but
+	// then it would be problematic to retrieve when the source is not available,
+	// and that's one of the reasons why we do this in the first place, so we can
+	// get the cached content even if source is not available.
+	//
+	// NOTE: We can use commit hash from Source and if it's not available use from
+	// lastAppliedRevision, lastAttemptedRevision, or lastPlannedRevision.
+	file, err := fs.GetWriter(fmt.Sprintf("%s-%s.tar.gz", terraform.GetNamespace(), terraform.GetName()))
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	for {
+		chunk, err := streamClient.Recv()
+		if err != nil {
+			if err == io.EOF {
+				if err := streamClient.CloseSend(); err != nil {
+					log.Error(err, "unabel to close stream")
+					break
+				}
+			}
+
+			return err
+		}
+
+		if len(chunk.Blob) > 0 {
+			if _, err := sha.Write(chunk.Blob); err != nil {
+				return err
+			}
+
+			if _, err := file.Write(chunk.Blob); err != nil {
+				return err
+			}
+		}
+
+		if len(chunk.Sha256Checksum) > 0 {
+			checksum = chunk.GetSha256Checksum()
+		}
+	}
+
+	log.Info("calculating checksum")
+	sum := sha.Sum(nil)
+
+	if !bytes.Equal(sum, checksum) {
+		return fmt.Errorf("invalid checksum, got: '%x'; expected: '%x'", sum, checksum)
+	}
+
+	return nil
+}

internal/storage/storage.go

@@ -22,7 +22,7 @@ func ArchiveDir(dir string) (out string, err error) {
 		}
 	}()
 
-	if err := os.Chdir(filepath.Dir(dir)); err != nil {
+	if err := os.Chdir(dir); err != nil {
 		return "", err
 	}
 

runner/runner.proto

@@ -33,14 +33,15 @@ service Runner {
 
   rpc Init(InitRequest) returns (InitReply) {}
   rpc SelectWorkspace(WorkspaceRequest) returns (WorkspaceReply) {}
-  rpc CreateWorkspaceBlob(CreateWorkspaceBlobRequest) returns (CreateWorkspaceBlobReply) {}
   rpc Upload(UploadRequest) returns (UploadReply) {}
 
   rpc FinalizeSecrets(FinalizeSecretsRequest) returns (FinalizeSecretsReply) {}
   rpc ForceUnlock(ForceUnlockRequest) returns (ForceUnlockReply) {}
 
   rpc StartBreakTheGlassSession(BreakTheGlassRequest) returns (BreakTheGlassReply) {}
   rpc HasBreakTheGlassSessionDone(BreakTheGlassRequest) returns (BreakTheGlassReply) {}
+
+  rpc CreateWorkspaceBlobStream(CreateWorkspaceBlobRequest) returns (stream CreateWorkspaceBlobReply) {}
 }
 
 message LookPathRequest {