[v2] Git source (#658)

Implements the git source for v2.

We currently only support secret refs from the Stack API because other
options (fs, env) don't make sense to share between agent and operator.
Auth information is passed via env vars to the agent's fetch container.

The operator also needs access to these secrets in order to pull for the
latest commit. It uses an in-memory `ls-remote` and compares remote
refs against's the Stack's ref. We support long and short forms (e.g.
`refs/heads/foo` or `foo`).

The agent now uses automation API to clone the repo, since it's a
natural place for dogfooding. If the init container is restarted, or uses
a persistent volume for the checkout, there's an edge case where
we won't fast-forward the existing checkout. See
pulumi/pulumi#17288.

Also adds a `shallow` clone option to the stack API for controlling
shallowness. We don't currently perform shallow clones so this is
backwards compatible.

Fixes #655

.github/workflows/v2-run-acceptance-tests.yaml

@@ -10,6 +10,8 @@ on:
 env:
   PULUMI_ACCESS_TOKEN: ${{ secrets.PULUMI_ACCESS_TOKEN }}
   GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+  PULUMI_BOT_TOKEN: ${{ secrets.PULUMI_BOT_TOKEN }}
+
 jobs:
   build:
     runs-on: ubuntu-latest
@@ -31,9 +33,9 @@ jobs:
           version: "~> v2"
           args: release --snapshot --clean --skip=docker
 
-  agent-integration-tests:
+  unit-tests:
     runs-on: ubuntu-latest
-    name: Integration Testing
+    name: Unit tests
     steps:
       - name: Check out code
         uses: actions/checkout@v4
@@ -55,3 +57,27 @@ jobs:
           files: agent/coverage.out,operator/coverage.out
         env:
           CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+
+  e2e-tests:
+    runs-on: ubuntu-latest
+    name: E2E tests
+    if: false
+    steps:
+      # Building the rootless image currently eats up all of our free disk.
+      - name: Free Disk Space (Ubuntu)
+        uses: jlumbroso/free-disk-space@main
+      - name: Setup cluster
+        uses: helm/kind-action@v1
+        with:
+          cluster_name: kind
+          node_image: kindest/node:v1.31.0
+      - name: Check out code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: 1.22.x
+      - name: Run tests
+        run: make -C operator test-e2e

.mise.toml

@@ -4,6 +4,7 @@ etcd = "3.5.16"
 protoc-gen-go = "1.34.2"
 protoc-gen-go-grpc = "1.5.1"
 kubebuilder = "4.2.0"
+envsubst = "1.4.2"
 
 [settings]
 experimental = true    # Enable Go backend.

agent/Makefile

@@ -11,7 +11,7 @@ protoc:
     	--go-grpc_out=. --go-grpc_opt=paths=source_relative *.proto
 
 test:
-	go test -covermode=atomic -coverprofile=coverage.out -v ./...
+	go test -covermode=atomic -coverprofile=coverage.out -coverpkg=./... -v ./...
 
 agent: protoc
 	@echo "Building agent"

agent/cmd/init.go

@@ -17,28 +17,25 @@ package cmd
 
 import (
 	"context"
-	"net/url"
+	"errors"
+	"fmt"
+	"io"
 	"os"
 
-	"github.com/fluxcd/pkg/git"
-	"github.com/fluxcd/pkg/git/gogit"
-	"github.com/fluxcd/pkg/git/repository"
+	"github.com/blang/semver"
 	"github.com/fluxcd/pkg/http/fetch"
-	"github.com/go-git/go-git/v5/plumbing/transport"
+	git "github.com/go-git/go-git/v5"
+	"github.com/pulumi/pulumi/sdk/v3/go/auto"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 )
 
-const (
-	DefaultFluxRetries = 3
-)
-
 var (
-	TargetDir   string
-	FluxUrl     string
-	FluxDigest  string
-	GitUrl      string
-	GitRevision string
+	_targetDir   string
+	_fluxURL     string
+	_fluxDigest  string
+	_gitURL      string
+	_gitRevision string
 )
 
 // initCmd represents the init command
@@ -50,99 +47,158 @@ var initCmd = &cobra.Command{
 For Flux sources:
 	pulumi-kubernetes-agent init --flux-fetch-url URL
 `,
-	Run: func(cmd *cobra.Command, args []string) {
+	RunE: func(cmd *cobra.Command, args []string) error {
 		ctx := cmd.Context()
-		log.Debugw("executing init command", "TargetDir", TargetDir)
+		var f fetchWithContexter
+		var g newLocalWorkspacer
 
-		err := os.MkdirAll(TargetDir, 0777)
-		if err != nil {
-			log.Errorw("fatal: unable to make target directory", zap.Error(err))
-			os.Exit(1)
-		}
-		log.Debugw("target directory created", "dir", TargetDir)
-
-		// fetch the configured flux artifact
-		if FluxUrl != "" {
-			// https://github.com/fluxcd/kustomize-controller/blob/a1a33f2adda783dd2a17234f5d8e84caca4e24e2/internal/controller/kustomization_controller.go#L328
-			fetcher := fetch.New(
-				fetch.WithRetries(DefaultFluxRetries),
-				fetch.WithHostnameOverwrite(os.Getenv("SOURCE_CONTROLLER_LOCALHOST")),
-				fetch.WithUntar())
-
-			log.Infow("flux artifact fetching", "url", FluxUrl, "digest", FluxDigest)
-			err := fetcher.FetchWithContext(ctx, FluxUrl, FluxDigest, TargetDir)
-			if err != nil {
-				log.Errorw("fatal: unable to fetch flux artifact", zap.Error(err))
-				os.Exit(2)
-			}
-			log.Infow("flux artifact fetched", "dir", TargetDir)
+		if _gitURL != "" {
+			g = &gitFetcher{url: _gitURL, revision: _gitRevision}
 		}
-
-		// fetch the configured git artifact
-		if GitUrl != "" {
-			u, err := url.Parse(GitUrl)
-			if err != nil {
-				log.Errorw("fatal: unable to parse git url", zap.Error(err))
-				os.Exit(2)
-			}
-			// Configure authentication strategy to access the source
-			authData := map[string][]byte{}
-			authOpts, err := git.NewAuthOptions(*u, authData)
-			if err != nil {
-				log.Errorw("fatal: unable to parse git auth options", zap.Error(err))
-				os.Exit(2)
-			}
-			cloneOpts := repository.CloneConfig{
-				RecurseSubmodules: false,
-				ShallowClone:      true,
+		// https://github.com/fluxcd/kustomize-controller/blob/a1a33f2adda783dd2a17234f5d8e84caca4e24e2/internal/controller/kustomization_controller.go#L328
+		if _fluxURL != "" {
+			f = &fluxFetcher{
+				url:    _fluxURL,
+				digest: _fluxDigest,
+				wrapped: fetch.New(
+					fetch.WithRetries(3),
+					fetch.WithHostnameOverwrite(os.Getenv("SOURCE_CONTROLLER_LOCALHOST")),
+					fetch.WithUntar()),
 			}
-			cloneOpts.Commit = GitRevision
-			log.Infow("git source fetching", "url", GitUrl, "revision", GitRevision)
-			_, err = gitCheckout(ctx, GitUrl, cloneOpts, authOpts, nil, TargetDir)
-			if err != nil {
-				log.Errorw("fatal: unable to fetch git source", zap.Error(err))
-				os.Exit(2)
-			}
-			log.Infow("git artifact fetched", "dir", TargetDir)
 		}
+		// Don't display usage on error.
+		cmd.SilenceUsage = true
+		return runInit(ctx, log, _targetDir, f, g)
 	},
 }
 
-func gitCheckout(ctx context.Context, url string, cloneOpts repository.CloneConfig,
-	authOpts *git.AuthOptions, proxyOpts *transport.ProxyOptions, dir string) (*git.Commit, error) {
+func runInit(ctx context.Context,
+	log *zap.SugaredLogger,
+	targetDir string,
+	f fetchWithContexter,
+	g newLocalWorkspacer,
+) error {
+	log.Debugw("executing init command", "TargetDir", targetDir)
+
+	// fetch the configured flux artifact
+	if f != nil {
+		err := os.MkdirAll(targetDir, 0o777)
+		if err != nil {
+			return fmt.Errorf("unable to make target directory: %w", err)
+		}
+		log.Debugw("target directory created", "dir", targetDir)
+
+		log.Infow("flux artifact fetching", "url", f.URL(), "digest", f.Digest())
+		err = f.FetchWithContext(ctx, f.URL(), f.Digest(), targetDir)
+		if err != nil {
+			return fmt.Errorf("unable to fetch flux artifact: %w", err)
+		}
+		log.Infow("flux artifact fetched", "dir", targetDir)
+		return nil
+	}
 
-	clientOpts := []gogit.ClientOption{gogit.WithDiskStorage()}
-	if authOpts.Transport == git.HTTP {
-		clientOpts = append(clientOpts, gogit.WithInsecureCredentialsOverHTTP())
+	// fetch the configured git artifact
+	auth := &auto.GitAuth{
+		SSHPrivateKey:       os.Getenv("GIT_SSH_PRIVATE_KEY"),
+		Username:            os.Getenv("GIT_USERNAME"),
+		Password:            os.Getenv("GIT_PASSWORD"),
+		PersonalAccessToken: os.Getenv("GIT_TOKEN"),
 	}
-	if proxyOpts != nil {
-		clientOpts = append(clientOpts, gogit.WithProxy(*proxyOpts))
+	repo := auto.GitRepo{
+		URL:        g.URL(),
+		CommitHash: g.Revision(),
+		Auth:       auth,
+		Shallow:    os.Getenv("GIT_SHALLOW") == "true",
 	}
 
-	gitReader, err := gogit.NewClient(dir, authOpts, clientOpts...)
-	if err != nil {
-		return nil, err
+	log.Infow("about to clone into", "TargetDir", targetDir, "fluxURL", g.URL())
+
+	// This will also handle creating the TargetDir for us.
+	_, err := g.NewLocalWorkspace(ctx,
+		auto.Repo(repo),
+		auto.WorkDir(targetDir),
+		auto.Pulumi(noop{}),
+	)
+	if errors.Is(err, git.ErrRepositoryAlreadyExists) {
+		// TODO(https://github.com/pulumi/pulumi/issues/17288): Automation
+		// API needs to ensure the existing checkout is valid.
+		log.Infow("repository was previously checked out", "dir", targetDir)
+		return nil
 	}
-	defer gitReader.Close()
-
-	commit, err := gitReader.Clone(ctx, url, cloneOpts)
 	if err != nil {
-		return nil, err
+		return fmt.Errorf("unable to fetch git source: %w", err)
 	}
+	log.Infow("git artifact fetched", "dir", targetDir)
+	return nil
+}
+
+type fetchWithContexter interface {
+	FetchWithContext(ctx context.Context, archiveURL, digest, dir string) error
+	URL() string
+	Digest() string
+}
+type fluxFetcher struct {
+	wrapped     *fetch.ArchiveFetcher
+	url, digest string
+}
+
+func (f fluxFetcher) FetchWithContext(ctx context.Context, archiveURL, digest, dir string) error {
+	return f.wrapped.FetchWithContext(ctx, archiveURL, digest, dir)
+}
+
+func (f fluxFetcher) URL() string {
+	return f.url
+}
+
+func (f fluxFetcher) Digest() string {
+	return f.digest
+}
+
+type newLocalWorkspacer interface {
+	NewLocalWorkspace(context.Context, ...auto.LocalWorkspaceOption) (auto.Workspace, error)
+	URL() string
+	Revision() string
+}
+
+type gitFetcher struct {
+	url, revision string
+}
+
+func (g gitFetcher) NewLocalWorkspace(ctx context.Context, opts ...auto.LocalWorkspaceOption) (auto.Workspace, error) {
+	return auto.NewLocalWorkspace(ctx, opts...)
+}
+
+func (g gitFetcher) URL() string {
+	return g.url
+}
 
-	return commit, nil
+func (g gitFetcher) Revision() string {
+	return g.revision
+}
+
+// noop is a PulumiCommand which doesn't do anything -- because the underlying CLI
+// isn't available in our image.
+type noop struct{}
+
+func (noop) Version() semver.Version { return semver.Version{} }
+
+func (noop) Run(context.Context, string, io.Reader, []io.Writer, []io.Writer, []string, ...string) (string, string, int, error) {
+	return "", "", 0, fmt.Errorf("pulumi CLI is not available during init")
 }
 
 func init() {
 	rootCmd.AddCommand(initCmd)
-	initCmd.Flags().StringVarP(&TargetDir, "target-dir", "t", "", "The target directory to initialize")
+	initCmd.Flags().StringVarP(&_targetDir, "target-dir", "t", "", "The target directory to initialize")
 	initCmd.MarkFlagRequired("target-dir")
 
-	initCmd.Flags().StringVar(&FluxUrl, "flux-url", "", "Flux archive URL")
-	initCmd.Flags().StringVar(&FluxDigest, "flux-digest", "", "Flux digest")
+	initCmd.Flags().StringVar(&_fluxURL, "flux-url", "", "Flux archive URL")
+	initCmd.Flags().StringVar(&_fluxDigest, "flux-digest", "", "Flux digest")
 	initCmd.MarkFlagsRequiredTogether("flux-url", "flux-digest")
 
-	initCmd.Flags().StringVar(&GitUrl, "git-url", "", "Git repository URL")
-	initCmd.Flags().StringVar(&GitRevision, "git-revision", "", "Git revision (tag or commit SHA)")
+	initCmd.Flags().StringVar(&_gitURL, "git-url", "", "Git repository URL")
+	initCmd.Flags().StringVar(&_gitRevision, "git-revision", "", "Git revision (tag or commit SHA)")
 	initCmd.MarkFlagsRequiredTogether("git-url", "git-revision")
+
+	initCmd.MarkFlagsOneRequired("git-url", "flux-url")
+	initCmd.MarkFlagsMutuallyExclusive("git-url", "flux-url")
 }