Add support for streaming logs from up/destroy/refresh (#101)

Also updates pulumi/pulumi dependency to v2.12.1.

Fixes #87.

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/operator-framework/operator-lib v0.0.0-20200728190837-b76db547798d
 	github.com/operator-framework/operator-sdk v0.19.0
 	github.com/pkg/errors v0.9.1
-	github.com/pulumi/pulumi/sdk/v2 v2.10.1-0.20200915174902-4e6ea760db2e
+	github.com/pulumi/pulumi/sdk/v2 v2.12.1
 	github.com/spf13/pflag v1.0.5
 	github.com/whilp/git-urls v1.0.0
 	golang.org/x/mod v0.3.0 // indirect

go.sum

@@ -787,10 +787,8 @@ github.com/prometheus/prometheus v0.0.0-20180315085919-58e2a31db8de/go.mod h1:oA
 github.com/prometheus/prometheus v1.8.2-0.20200110114423-1e64d757f711/go.mod h1:7U90zPoLkWjEIQcy/rweQla82OCTUzxVHE51G3OhJbI=
 github.com/prometheus/prometheus v2.3.2+incompatible/go.mod h1:oAIUtOny2rjMX0OWN5vPR5/q/twIROJvdqnQKDdil/s=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
-github.com/pulumi/pulumi/sdk/v2 v2.10.1-0.20200914235159-89b65ebd443d h1:Vq+2yJNR8wiZCMZe8CitnhQysHBcevkw51cz8HoCEec=
-github.com/pulumi/pulumi/sdk/v2 v2.10.1-0.20200914235159-89b65ebd443d/go.mod h1:EED7KCDOohYIewUppsav5KHTFTmfYGqUFib1uRvYdWQ=
-github.com/pulumi/pulumi/sdk/v2 v2.10.1-0.20200915174902-4e6ea760db2e h1:Gvc3rEk40o70cFK06HlV0icHk0SlM10E4VXTuqHjMHA=
-github.com/pulumi/pulumi/sdk/v2 v2.10.1-0.20200915174902-4e6ea760db2e/go.mod h1:EED7KCDOohYIewUppsav5KHTFTmfYGqUFib1uRvYdWQ=
+github.com/pulumi/pulumi/sdk/v2 v2.12.1 h1:rEQHyjaGSGybqqeKLMJZH034UemMPGw2AWzpGcn1tF4=
+github.com/pulumi/pulumi/sdk/v2 v2.12.1/go.mod h1:WQ4WaHMA7mduVHAJi87iIqbWvqsuBUYccBiKK+FrayI=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/remyoudompheng/bigfft v0.0.0-20170806203942-52369c62f446/go.mod h1:uYEyJGbgTkfkS4+E/PavXkNJcbFIpEtjt2B0KDQ5+9M=
 github.com/robfig/cron v0.0.0-20170526150127-736158dc09e1/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=

pkg/controller/stack/stack_controller.go

@@ -20,7 +20,9 @@ import (
 	"github.com/pkg/errors"
 	pulumiv1alpha1 "github.com/pulumi/pulumi-kubernetes-operator/pkg/apis/pulumi/v1alpha1"
 	"github.com/pulumi/pulumi/sdk/v2/go/x/auto"
+	"github.com/pulumi/pulumi/sdk/v2/go/x/auto/optdestroy"
 	"github.com/pulumi/pulumi/sdk/v2/go/x/auto/optrefresh"
+	"github.com/pulumi/pulumi/sdk/v2/go/x/auto/optup"
 	giturls "github.com/whilp/git-urls"
 	git "gopkg.in/src-d/go-git.v4"
 	"gopkg.in/src-d/go-git.v4/plumbing"
@@ -627,8 +629,11 @@ func (sess *reconcileStackSession) UpdateConfig() error {
 }
 
 func (sess *reconcileStackSession) RefreshStack(expectNoChanges bool) (pulumiv1alpha1.Permalink, error) {
-	result, err := sess.autoStack.Refresh(context.Background(),
+	writer := logWriter(sess.logger, "Pulumi Refresh")
+	result, err := sess.autoStack.Refresh(
+		context.Background(),
 		optrefresh.ExpectNoChanges(),
+		optrefresh.ProgressStreams(writer),
 	)
 	if err != nil {
 		return pulumiv1alpha1.Permalink(""), errors.Wrapf(err, "refreshing stack '%s'", sess.stack.Stack)
@@ -645,7 +650,8 @@ func (sess *reconcileStackSession) RefreshStack(expectNoChanges bool) (pulumiv1a
 // and error. In certain cases, an update may be unabled to proceed due to locking,
 // in which case the operator will requeue itself to retry later.
 func (sess *reconcileStackSession) UpdateStack() (pulumiv1alpha1.StackUpdateStatus, pulumiv1alpha1.Permalink, *auto.UpResult, error) {
-	result, err := sess.autoStack.Up(context.Background())
+	writer := logWriter(sess.logger, "Pulumi Update")
+	result, err := sess.autoStack.Up(context.Background(), optup.ProgressStreams(writer))
 	if err != nil {
 		// If this is the "conflict" error message, we will want to gracefully quit and retry.
 		if auto.IsConcurrentUpdateError(err) {
@@ -684,7 +690,8 @@ func (sess *reconcileStackSession) GetStackOutputs(outs auto.OutputMap) (pulumiv
 }
 
 func (sess *reconcileStackSession) DestroyStack() error {
-	_, err := sess.autoStack.Destroy(context.Background())
+	writer := logWriter(sess.logger, "Pulumi Destroy")
+	_, err := sess.autoStack.Destroy(context.Background(), optdestroy.ProgressStreams(writer))
 	if err != nil {
 		return errors.Wrapf(err, "destroying resources for stack '%s'", sess.stack.Stack)
 	}
@@ -959,3 +966,20 @@ func waitForFile(fp string) ([]byte, error) {
 	}
 	return file, err
 }
+
+// logWriter constructs an io.Writer that logs to the provided logr.Logger
+func logWriter(logger logr.Logger, msg string, keysAndValues ...interface{}) io.Writer {
+	stdoutR, stdoutW := io.Pipe()
+	go func() {
+		outs := bufio.NewScanner(stdoutR)
+		for outs.Scan() {
+			text := outs.Text()
+			logger.Info(msg, append([]interface{}{"Stdout", text}, keysAndValues...)...)
+		}
+		err := outs.Err()
+		if err != nil {
+			logger.Error(err, msg, keysAndValues...)
+		}
+	}()
+	return stdoutW
+}