implement local gRPC server for terraform runner

Dockerfile

@@ -12,12 +12,12 @@ COPY go.sum go.sum
 RUN go mod download
 
 # Copy the go source
-COPY main.go main.go
+COPY cmd/manager/main.go cmd/manager/main.go
 COPY api/ api/
 COPY controllers/ controllers/
 
 # Build
-RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -o tf-controller main.go
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -o tf-controller cmd/controller/main.go
 
 ADD https://releases.hashicorp.com/terraform/1.1.4/terraform_1.1.4_linux_amd64.zip /terraform_1.1.4_linux_amd64.zip
 RUN unzip -q /terraform_1.1.4_linux_amd64.zip

Makefile

@@ -68,17 +68,20 @@ download-crd-deps:
 
 .PHONY: test
 test: manifests generate download-crd-deps fmt vet envtest api-docs ## Run tests.
-	DISABLE_K8S_LOGS=1 DISABLE_TF_LOGS=1 DISABLE_TF_K8S_BACKEND=1 KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) -p path)" go test ./controllers -coverprofile cover.out -v
+	INSECURE_LOCAL_RUNNER=1 DISABLE_K8S_LOGS=1 DISABLE_TF_LOGS=1 DISABLE_TF_K8S_BACKEND=1 KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) -p path)" go test ./controllers -coverprofile cover.out -v
 
+gen-grpc:
+	protoc --go_out=. --go_opt=Mrunner/runner.proto=runner/ --go-grpc_out=. --go-grpc_opt=Mrunner/runner.proto=runner/ runner/runner.proto
 ##@ Build
 
 .PHONY: build
-build: generate fmt vet ## Build manager binary.
-	go build -o bin/manager main.go
+build: gen-grpc generate fmt vet ## Build manager binary.
+	go build -o bin/runner cmd/runner/main.go
+	go build -o bin/manager cmd/manager/main.go
 
 .PHONY: run
 run: manifests generate fmt vet ## Run a controller from your host.
-	go run ./main.go
+	go run cmd/manager/main.go
 
 .PHONY: docker-build
 docker-build: test ## Build docker image with the manager.

api/v1alpha1/terraform_types.go

@@ -18,6 +18,9 @@ package v1alpha1
 
 import (
 	"fmt"
+	sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"strings"
 	"time"
 
@@ -177,12 +180,12 @@ type TerraformStatus struct {
 
 	// LastDriftDetectedAt is the time when the last drift was detected
 	// +optional
-	LastDriftDetectedAt metav1.Time `json:"lastDriftDetectedAt,omitempty"`
+	LastDriftDetectedAt *metav1.Time `json:"lastDriftDetectedAt,omitempty"`
 
 	// LastAppliedByDriftDetectionAt is the time when the last drift was detected and
 	// terraform apply was performed as a result
 	// +optional
-	LastAppliedByDriftDetectionAt metav1.Time `json:"lastAppliedByDriftDetectionAt,omitempty"`
+	LastAppliedByDriftDetectionAt *metav1.Time `json:"lastAppliedByDriftDetectionAt,omitempty"`
 
 	// +optional
 	AvailableOutputs []string `json:"availableOutputs,omitempty"`
@@ -294,7 +297,7 @@ func TerraformApplied(terraform Terraform, revision string, message string, isDe
 	meta.SetResourceCondition(&terraform, "Apply", metav1.ConditionTrue, TFExecApplySucceedReason, message)
 
 	if terraform.Status.Plan.IsDriftDetectionPlan {
-		(&terraform).Status.LastAppliedByDriftDetectionAt = metav1.Now()
+		(&terraform).Status.LastAppliedByDriftDetectionAt = &metav1.Time{Time: time.Now()}
 	}
 
 	(&terraform).Status.Plan = PlanStatus{
@@ -377,7 +380,7 @@ func TerraformAppliedFailResetPlanAndNotReady(terraform Terraform, revision, rea
 }
 
 func TerraformDriftDetected(terraform Terraform, revision, reason, message string) Terraform {
-	(&terraform).Status.LastDriftDetectedAt = metav1.Now()
+	(&terraform).Status.LastDriftDetectedAt = &metav1.Time{Time: time.Now()}
 	SetTerraformReadiness(&terraform, metav1.ConditionFalse, reason, trimString(message, MaxConditionMessageLength), revision)
 	return terraform
 }
@@ -402,7 +405,8 @@ func (in Terraform) HasDrift() bool {
 	for _, condition := range in.Status.Conditions {
 		if condition.Type == "Apply" &&
 			condition.Status == metav1.ConditionTrue &&
-			in.Status.LastDriftDetectedAt.After(condition.LastTransitionTime.Time) {
+			in.Status.LastDriftDetectedAt != nil &&
+			(*in.Status.LastDriftDetectedAt).After(condition.LastTransitionTime.Time) {
 			return true
 		}
 	}
@@ -422,6 +426,24 @@ func (in *Terraform) GetStatusConditions() *[]metav1.Condition {
 	return &in.Status.Conditions
 }
 
+func (in Terraform) ToBytes(scheme *runtime.Scheme) ([]byte, error) {
+	return runtime.Encode(
+		serializer.NewCodecFactory(scheme).LegacyCodec(
+			corev1.SchemeGroupVersion,
+			GroupVersion,
+			sourcev1.GroupVersion,
+		), &in)
+}
+
+func (in *Terraform) FromBytes(b []byte, scheme *runtime.Scheme) error {
+	return runtime.DecodeInto(
+		serializer.NewCodecFactory(scheme).LegacyCodec(
+			corev1.SchemeGroupVersion,
+			GroupVersion,
+			sourcev1.GroupVersion,
+		), b, in)
+}
+
 func trimString(str string, limit int) string {
 	if len(str) <= limit {
 		return str

main.go → cmd/manager/main.go

@@ -17,6 +17,9 @@ limitations under the License.
 package main
 
 import (
+	"github.com/chanwit/tf-controller/runner"
+	"google.golang.org/grpc"
+	"net"
 	"os"
 	"time"
 
@@ -147,6 +150,26 @@ func main() {
 		os.Exit(1)
 	}
 
+	if os.Getenv("INSECURE_LOCAL_RUNNER") == "1" {
+		listener, err := net.Listen("tcp", "localhost:30000")
+		if err != nil {
+			panic(err.Error())
+		}
+
+		server := grpc.NewServer()
+		// local runner, use the same client as the manager
+		runner.RegisterRunnerServer(server, &runner.TerraformRunnerServer{
+			Client: mgr.GetClient(),
+			Scheme: mgr.GetScheme(),
+		})
+
+		go func() {
+			if err := server.Serve(listener); err != nil {
+				panic(err.Error())
+			}
+		}()
+	}
+
 	setupLog.Info("starting manager")
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		setupLog.Error(err, "problem running manager")

cmd/runner/main.go

@@ -0,0 +1,26 @@
+/*
+Copyright 2021.
+
+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 main
+
+import (
+	"fmt"
+	_ "github.com/chanwit/tf-controller/runner"
+)
+
+func main() {
+	fmt.Println("hello world")
+}

controllers/object_encode_test.go

@@ -0,0 +1,41 @@
+package controllers
+
+import (
+	infrav1 "github.com/chanwit/tf-controller/api/v1alpha1"
+	. "github.com/onsi/gomega"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"testing"
+)
+
+func TestObjectEncode(t *testing.T) {
+	g := NewGomegaWithT(t)
+
+	helloWorldTF := infrav1.Terraform{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: infrav1.GroupVersion.Group + "/" + infrav1.GroupVersion.Version,
+			Kind:       infrav1.TerraformKind,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "tf-1",
+			Namespace: "flux-system",
+		},
+		Spec: infrav1.TerraformSpec{
+			ApprovePlan: "auto",
+			Path:        "./terraform-hello-world-example",
+			SourceRef: infrav1.CrossNamespaceSourceReference{
+				Kind:      "GitRepository",
+				Name:      "flux-system",
+				Namespace: "flux-system",
+			},
+		},
+	}
+
+	b, err := helloWorldTF.ToBytes(reconciler.Scheme)
+	g.Expect(err).To(BeNil())
+	var tf infrav1.Terraform
+	err = tf.FromBytes(b, runnerServer.Scheme)
+	g.Expect(err).To(BeNil())
+	g.Expect(tf).To(Equal(helloWorldTF))
+	g.Expect(tf.Spec.ApprovePlan).To(Equal("auto"))
+	g.Expect(tf.Spec.Path).To(Equal("./terraform-hello-world-example"))
+}

controllers/suite_test.go

@@ -19,9 +19,14 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"github.com/chanwit/tf-controller/runner"
+	"google.golang.org/grpc"
 	"io"
 	"io/ioutil"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"math/rand"
+	"net"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -36,7 +41,6 @@ import (
 
 	infrav1 "github.com/chanwit/tf-controller/api/v1alpha1"
 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
-	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -58,11 +62,12 @@ const (
 )
 
 var (
-	cfg        *rest.Config
-	k8sClient  client.Client
-	testEnv    *envtest.Environment
-	server     *ghttp.Server
-	reconciler *TerraformReconciler
+	cfg          *rest.Config
+	k8sClient    client.Client
+	testEnv      *envtest.Environment
+	server       *ghttp.Server
+	reconciler   *TerraformReconciler
+	runnerServer *runner.TerraformRunnerServer
 )
 
 var (
@@ -97,18 +102,25 @@ func TestMain(m *testing.M) {
 		panic("cfg cannot be nil")
 	}
 
-	err = sourcev1.AddToScheme(scheme.Scheme)
+	scheme := runtime.NewScheme()
+
+	err = clientgoscheme.AddToScheme(scheme)
+	if err != nil {
+		panic(err.Error())
+	}
+
+	err = sourcev1.AddToScheme(scheme)
 	if err != nil {
 		panic(err.Error())
 	}
 
-	err = infrav1.AddToScheme(scheme.Scheme)
+	err = infrav1.AddToScheme(scheme)
 	if err != nil {
 		panic(err.Error())
 	}
 
 	//+kubebuilder:scaffold:scheme
-	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
+	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme})
 	if err != nil {
 		panic(err.Error())
 	}
@@ -166,7 +178,7 @@ func TestMain(m *testing.M) {
 	}
 
 	k8sManager, err := ctrl.NewManager(cfg, ctrl.Options{
-		Scheme: scheme.Scheme,
+		Scheme: scheme,
 	})
 	if err != nil {
 		panic(err.Error())
@@ -185,8 +197,24 @@ func TestMain(m *testing.M) {
 	}
 
 	go func() {
-		err = k8sManager.Start(ctx)
-		if err != nil {
+		if err := k8sManager.Start(ctx); err != nil {
+			panic(err.Error())
+		}
+	}()
+
+	listener, err := net.Listen("tcp", "localhost:30000")
+	if err != nil {
+		panic(err.Error())
+	}
+	s := grpc.NewServer()
+	runnerServer = &runner.TerraformRunnerServer{
+		Client: k8sManager.GetClient(),
+		Scheme: scheme,
+	}
+	runner.RegisterRunnerServer(s, runnerServer)
+
+	go func() {
+		if err := s.Serve(listener); err != nil {
 			panic(err.Error())
 		}
 	}()

controllers/tc000011_bad_tar_gz_no_outputs_test.go

@@ -137,7 +137,7 @@ func Test_000011_bad_tar_gz_no_outputs_test(t *testing.T) {
 	}, timeout, interval).Should(Equal(map[string]interface{}{
 		"Type":    "Ready",
 		"Reason":  "ArtifactFailed",
-		"Message": "failed to untar artifact, error: requires gzip-compressed body: gzip: invalid header",
+		"Message": "rpc error: code = Unknown desc = failed to untar artifact, error: requires gzip-compressed body: gzip: invalid header",
 		"Status":  metav1.ConditionFalse,
 	}))
 }

controllers/tc000072_varsfrom_optional_secret_controlled_outputs_test.go

@@ -123,7 +123,7 @@ func Test_000072_varsfrom_optional_secret_and_controlled_outputs_test(t *testing
 		return helloWorldTF.Status
 	}, timeout, interval).Should(Equal(map[string]interface{}{
 		"Reason":  "VarsGenerationFailed",
-		"Message": "Secret \"my-vars-helloworld-vars-from-optional-secret-controlled-output\" not found",
+		"Message": "rpc error: code = Unknown desc = Secret \"my-vars-helloworld-vars-from-optional-secret-controlled-output\" not found",
 		"Status":  metav1.ConditionFalse,
 		"Type":    "Ready",
 	}))

controllers/tc000074_varsfrom_accepts_many_secrets_with_last_supplied_key_precedence_test.go

@@ -3,6 +3,7 @@ package controllers
 import (
 	"context"
 	"encoding/json"
+	"github.com/chanwit/tf-controller/runner"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -90,7 +91,13 @@ func Test_000074_varsfrom_accepts_many_secrets_with_last_supplied_key_precedence
 		},
 	}
 
-	_, err = reconciler.generateVarsForTF(ctx, terraform, tfExec, "main")
+	terraformBytes, err := terraform.ToBytes(reconciler.Scheme)
+	g.Expect(err).To(BeNil())
+
+	_, err = runnerServer.GenerateVarsForTF(ctx, &runner.GenerateVarsForTFRequest{
+		WorkingDir: tfExec.WorkingDir(),
+		Terraform:  terraformBytes,
+	})
 	g.Expect(err).Should(BeNil())
 
 	By("verifying the generated vars file matches the expected result")