Check Flux sources for artifact to avoid spamming the log (#359)

* Check for an artifact in Flux sources

When a Stack refers to a Flux source, we check the latter for a `Ready`
condition, and if present and false, bail on reconciliation until it
gets requeued by the watch mechanism.

A flaw in this is that a new source will have no Ready condition (true
or false), so the controller will continue, only to find there's no
record of what it should download. Instead of gracefully parking itself,
it bounces off this until the source has been seen by the
source-controller.

So: check that there's an artifact.

* Use github.com/fluxcd/pkg/http/fetch

This obviates a bunch of my own code (and is a bit more careful with
buffers).

* Add changelog entry

Signed-off-by: Michael Bridgen <mbridgen@pulumi.com>

CHANGELOG.md

@@ -14,6 +14,9 @@ CHANGELOG
   [#369](https://github.com/pulumi/pulumi-kubernetes-operator/pull/369)
 - Use an init process so processes spawned by `pulumi` are reaped
   [#367](https://github.com/pulumi/pulumi-kubernetes-operator/pull/367)
+- When a Stack uses a Flux source, but the source has no artifact to download, park the Stack until
+  the source has been updated, rather than retrying
+  [#359](https://github.com/pulumi/pulumi-kubernetes-operator/pull/359)
 
 ## 1.10.1 (2022-10-25)
 

go.mod

@@ -40,6 +40,7 @@ require (
 )
 
 require (
+	github.com/fluxcd/pkg/http/fetch v0.2.0
 	github.com/go-git/go-git/v5 v5.4.2
 	github.com/onsi/ginkgo/v2 v2.3.1
 	sigs.k8s.io/yaml v1.2.0
@@ -61,10 +62,12 @@ require (
 	github.com/blang/semver v3.5.1+incompatible // indirect
 	github.com/cheggaaa/pb v1.0.18 // indirect
 	github.com/coreos/prometheus-operator v0.38.1-0.20200424145508-7e176fda06cc // indirect
+	github.com/cyphar/filepath-securejoin v0.2.3 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/djherbis/times v1.2.0 // indirect
 	github.com/emirpasic/gods v1.12.0 // indirect
 	github.com/evanphx/json-patch v4.11.0+incompatible // indirect
+	github.com/fluxcd/pkg/tar v0.2.0 // indirect
 	github.com/form3tech-oss/jwt-go v3.2.2+incompatible // indirect
 	github.com/go-git/gcfg v1.5.0 // indirect
 	github.com/go-git/go-billy/v5 v5.3.1 // indirect
@@ -78,7 +81,9 @@ require (
 	github.com/googleapis/gnostic v0.5.5 // indirect
 	github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645 // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-cleanhttp v0.5.1 // indirect
 	github.com/hashicorp/go-multierror v1.0.0 // indirect
+	github.com/hashicorp/go-retryablehttp v0.7.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect

go.sum

@@ -223,6 +223,8 @@ github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7Do
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/cyphar/filepath-securejoin v0.2.2/go.mod h1:FpkQEhXnPnOthhzymB7CGsFk2G9VLXONKD9G7QGMM+4=
+github.com/cyphar/filepath-securejoin v0.2.3 h1:YX6ebbZCZP7VkM3scTTokDgBL2TY741X51MTk3ycuNI=
+github.com/cyphar/filepath-securejoin v0.2.3/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4=
 github.com/cznic/b v0.0.0-20180115125044-35e9bbe41f07/go.mod h1:URriBxXwVq5ijiJ12C7iIZqlA69nTlI+LgI6/pwftG8=
 github.com/cznic/fileutil v0.0.0-20180108211300-6a051e75936f/go.mod h1:8S58EK26zhXSxzv7NQFpnliaOQsmDUxvoQO3rt154Vg=
 github.com/cznic/golex v0.0.0-20170803123110-4ab7c5e190e4/go.mod h1:+bmmJDNmKlhWNG+gwWCkaBoTy39Fs+bzRxVBzoTQbIc=
@@ -306,6 +308,12 @@ github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL
 github.com/fatih/color v1.12.0 h1:mRhaKNwANqRgUBGKmnI5ZxEk7QXmjQeCcuYFMX2bfcc=
 github.com/fatih/color v1.12.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
 github.com/fatih/structtag v1.1.0/go.mod h1:mBJUNpUnHmRKrKlQQlmCrh5PuhftFbNv8Ys4/aAZl94=
+github.com/fluxcd/pkg/http/fetch v0.2.0 h1:Ss2bPfEn1e7OLebzkrU2c2bT1vZWFZmHCjnl0ACsRYM=
+github.com/fluxcd/pkg/http/fetch v0.2.0/go.mod h1:60QOWiz4pLe8SPtlHZtVo92qga18qQT2PrbU0D5NWqM=
+github.com/fluxcd/pkg/tar v0.2.0 h1:HEUHgONQYsJGeZZ4x6h5nQU9Aox1I4T3bOp1faWTqf8=
+github.com/fluxcd/pkg/tar v0.2.0/go.mod h1:w0/TOC7kwBJhnSJn7TCABkc/I7ib1f2Yz6vOsbLBnhw=
+github.com/fluxcd/pkg/testserver v0.3.0 h1:oyZW6YWHVZR7FRVNu7lN9F5H808TD2jCzBm8CenFoi0=
+github.com/fluxcd/pkg/testserver v0.3.0/go.mod h1:gjOKX41okmrGYOa4oOF2fiLedDAfPo1XaG/EzrUUGBI=
 github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:xEzjJPgXI435gkrCt3MPfRiAkVrwSbHsst4LCFVfpJc=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible h1:TcekIExNqud5crz4xD2pavyTgWiPvpYe4Xau31I0PRk=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
@@ -569,7 +577,10 @@ github.com/hashicorp/errwrap v0.0.0-20141028054710-7554cd9344ce/go.mod h1:YH+1FK
 github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
+github.com/hashicorp/go-cleanhttp v0.5.1 h1:dH3aiDG9Jvb5r5+bYHsikaOUIpcM0xvgMXVoDkXMzJM=
 github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
+github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI=
+github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
 github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
 github.com/hashicorp/go-immutable-radix v1.1.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
 github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
@@ -578,6 +589,8 @@ github.com/hashicorp/go-multierror v0.0.0-20161216184304-ed905158d874/go.mod h1:
 github.com/hashicorp/go-multierror v1.0.0 h1:iVjPR7a6H0tWELX5NxNe7bYopibicUzc7uPribsnS6o=
 github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
 github.com/hashicorp/go-retryablehttp v0.5.3/go.mod h1:9B5zBasrRhHXnJnui7y6sL7es7NDiJgTc6Er0maI1Xs=
+github.com/hashicorp/go-retryablehttp v0.7.1 h1:sUiuQAnLlbvmExtFQs72iFW/HXeUn8Z1aJLQ4LJJbTQ=
+github.com/hashicorp/go-retryablehttp v0.7.1/go.mod h1:vAew36LZh98gCBJNLH42IQ1ER/9wtLZZ8meHqQvEYWY=
 github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU=
 github.com/hashicorp/go-rootcerts v1.0.1/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR319Vf4pU3K5EGc8=
 github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
@@ -812,6 +825,7 @@ github.com/onsi/gomega v1.13.0/go.mod h1:lRk9szgn8TxENtWd0Tp4c3wjlRfMTMH27I+3Je4
 github.com/onsi/gomega v1.17.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY=
 github.com/onsi/gomega v1.19.0/go.mod h1:LY+I3pBVzYsTBU1AnDwOSxaYi9WoWiqgwooUqq9yPro=
 github.com/onsi/gomega v1.20.1/go.mod h1:DtrZpjmvpn2mPm4YWQa0/ALMDj9v4YxLgojwPeREyVo=
+github.com/onsi/gomega v1.21.1/go.mod h1:iYAIXgPSaDHak0LCMA+AWBpIKBr8WZicMxnE8luStNc=
 github.com/onsi/gomega v1.22.0 h1:AIg2/OntwkBiCg5Tt1ayyiF1ArFrWFoCSMtMi/wdApk=
 github.com/onsi/gomega v1.22.0/go.mod h1:iYAIXgPSaDHak0LCMA+AWBpIKBr8WZicMxnE8luStNc=
 github.com/opencontainers/go-digest v0.0.0-20170106003457-a6d0ee40d420/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=

pkg/controller/stack/flux.go

@@ -3,22 +3,21 @@
 package stack
 
 import (
-	"bytes"
 	"context"
-	"crypto/sha1"
-	"crypto/sha256"
 	"fmt"
-	"io"
-	"net/http"
 	"os"
 	"path/filepath"
 
+	"github.com/fluxcd/pkg/http/fetch"
+
 	"github.com/pulumi/pulumi/sdk/v3/go/auto"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 
 	"github.com/pulumi/pulumi-kubernetes-operator/pkg/apis/pulumi/shared"
 )
 
+const maxArtifactDownloadSize = 50 * 1024 * 1024
+
 func (sess *reconcileStackSession) SetupWorkdirFromFluxSource(ctx context.Context, source unstructured.Unstructured, fluxSource *shared.FluxSource) (_commit string, retErr error) {
 	rootdir, err := os.MkdirTemp("", "pulumi_source")
 	if err != nil {
@@ -57,38 +56,9 @@ func (sess *reconcileStackSession) SetupWorkdirFromFluxSource(ctx context.Contex
 		return "", err
 	}
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, artifactURL, nil)
-	if err != nil {
-		return "", fmt.Errorf("failed to create a request: %w", err)
-	}
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		return "", fmt.Errorf("request for artifact failed: %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return "", fmt.Errorf("failed to download artifact from %s, status %q (expected 200 OK)", artifactURL, resp.Status)
-	}
-	// TODO validate size, if given
-
-	defer resp.Body.Close()
-
-	var buf bytes.Buffer
-	hasher := sha256.New()
-	if len(checksum) == 40 { // Flux source-controller <= 0.17.2 used SHA1
-		hasher = sha1.New()
-	}
-	out := io.MultiWriter(hasher, &buf)
-	if _, err := io.Copy(out, resp.Body); err != nil {
-		return "", fmt.Errorf("failed to compute checksum from artifact response: %w", err)
-	}
-	if checksum1 := fmt.Sprintf("%x", hasher.Sum(nil)); checksum1 != checksum {
-		return "", fmt.Errorf("computed checksum of artifact %q does not match checksum recorded %q", checksum1, checksum)
-	}
-
-	// we downloaded the artifact gzip-tarball into a buffer and it matches the checksum; untar it
-	// into our working dir.
-	if err = untar(&buf, rootdir); err != nil {
-		return "", fmt.Errorf("failed to extract archive tarball: %w", err)
+	fetcher := fetch.NewArchiveFetcher(1, maxArtifactDownloadSize, maxArtifactDownloadSize*10, "")
+	if err = fetcher.Fetch(artifactURL, checksum, rootdir); err != nil {
+		return "", fmt.Errorf("failed to get artifact from source: %w", err)
 	}
 
 	// woo! now there's a directory with source in `rootdir`. Construct a workspace.
@@ -107,25 +77,30 @@ func (sess *reconcileStackSession) SetupWorkdirFromFluxSource(ctx context.Contex
 // ready, and nil if it cannot determine so.
 func checkFluxSourceReady(obj unstructured.Unstructured) error {
 	conditions, ok, err := unstructured.NestedSlice(obj.Object, "status", "conditions")
-	if err != nil || !ok {
-		// didn't find a []Condition, so there's nothing to indicate that it's not ready
-		return nil
-	}
-	for _, c0 := range conditions {
-		var c map[string]interface{}
-		if c, ok = c0.(map[string]interface{}); !ok {
-			// condition isn't the right shape, move on
-			continue
-		}
-		if t, ok, err := unstructured.NestedString(c, "type"); ok && err == nil && t == "Ready" {
-			if v, ok, err := unstructured.NestedString(c, "status"); ok && err == nil && v == "True" {
-				// found the Ready condition and it is actually ready
-				return nil
+	if ok && err == nil {
+		// didn't find a []Condition, so there's nothing to indicate that it's not ready there
+		for _, c0 := range conditions {
+			var c map[string]interface{}
+			if c, ok = c0.(map[string]interface{}); !ok {
+				// condition isn't the right shape, try the next one
+				continue
+			}
+			if t, ok, err := unstructured.NestedString(c, "type"); ok && err == nil && t == "Ready" {
+				if v, ok, err := unstructured.NestedString(c, "status"); ok && err == nil && v == "True" {
+					// found the Ready condition and it is actually ready; proceed to next check
+					break
+				}
+				// found the Ready condition and it's something other than ready
+				return fmt.Errorf("source Ready condition does not have status True %#v", c)
 			}
-			// found the Ready condition and it's something other than ready
-			return fmt.Errorf("source Ready condition does not have status True %#v", c)
 		}
+		// Ready=true, or no ready condition to tell us either way
+	}
+
+	_, ok, err = unstructured.NestedMap(obj.Object, "status", "artifact")
+	if !ok || err != nil {
+		return fmt.Errorf(".status.artifact does not have an Artifact object")
 	}
-	// didn't find the Ready condition
+
 	return nil
 }

test/flux_source_test.go

@@ -338,6 +338,34 @@ var _ = Describe("Flux source integration", func() {
 			})
 		})
 
+		When("the source object has no artifact", func() {
+			BeforeEach(func() {
+				unstructured.RemoveNestedField(source.Object, "status", "artifact")
+				Expect(k8sClient.Status().Update(context.TODO(), source)).To(Succeed())
+				stack.Name = "source-no-artifact"
+			})
+
+			It("marks the stack as failed and to be retried", func() {
+				waitForStackFailure(stack)
+				expectInProgress(stack.Status.Conditions)
+
+				newArtifactRevision := randString()
+				By("putting the artifact in the status, the stack can run")
+				artifact := map[string]interface{}{
+					"path":     "irrelevant",
+					"url":      artifactURL,
+					"revision": newArtifactRevision,
+					"checksum": artifactChecksum,
+				}
+				unstructured.SetNestedMap(source.Object, artifact, "status", "artifact")
+				Expect(k8sClient.Status().Update(context.TODO(), source)).To(Succeed())
+
+				waitForStackSuccess(stack)
+				expectReady(stack.Status.Conditions)
+				Expect(stack.Status.LastUpdate.LastSuccessfulCommit).To(Equal(newArtifactRevision))
+			})
+		})
+
 		When("the checksum is wrong", func() {
 			BeforeEach(func() {
 				unstructured.SetNestedField(source.Object, "not-the-right-checksum",