Merge pull request #219 from fluxcd/debug-wait-logs

controllers/helmrelease_controller.go

@@ -328,14 +328,14 @@ func (r *HelmReleaseReconciler) reconcileRelease(ctx context.Context,
 		// Fail if install retries are exhausted.
 		if hr.Spec.GetInstall().GetRemediation().RetriesExhausted(hr) {
 			err = fmt.Errorf("install retries exhausted")
-			return v2.HelmReleaseNotReady(hr, released.Reason, released.Message), err
+			return v2.HelmReleaseNotReady(hr, released.Reason, err.Error()), err
 		}
 
 		// Fail if there is a release and upgrade retries are exhausted.
 		// This avoids failing after an upgrade uninstall remediation strategy.
 		if rel != nil && hr.Spec.GetUpgrade().GetRemediation().RetriesExhausted(hr) {
 			err = fmt.Errorf("upgrade retries exhausted")
-			return v2.HelmReleaseNotReady(hr, released.Reason, released.Message), err
+			return v2.HelmReleaseNotReady(hr, released.Reason, err.Error()), err
 		}
 	}
 
@@ -415,9 +415,8 @@ func (r *HelmReleaseReconciler) reconcileRelease(ctx context.Context,
 
 	if err != nil {
 		reason := meta.ReconciliationFailedReason
-		var cerr *ConditionError
+		if condErr := (*ConditionError)(nil); errors.As(err, &condErr) {
-		if errors.As(err, &cerr) {
+			reason = condErr.Reason
-			reason = cerr.Reason
 		}
 		return v2.HelmReleaseNotReady(hr, reason, err.Error()), err
 	}
@@ -662,10 +661,14 @@ func (r *HelmReleaseReconciler) reconcileDelete(ctx context.Context, hr v2.HelmR
 func (r *HelmReleaseReconciler) handleHelmActionResult(ctx context.Context,
 	hr *v2.HelmRelease, revision string, err error, action string, condition string, succeededReason string, failedReason string) error {
 	if err != nil {
-		msg := fmt.Sprintf("Helm %s failed: %s", action, err.Error())
+		err = fmt.Errorf("Helm %s failed: %w", action, err)
+		msg := err.Error()
+		if actionErr := (*runner.ActionError)(nil); errors.As(err, &actionErr) {
+			msg = msg + "\n\nLast Helm logs:\n\n" + actionErr.CapturedLogs
+		}
 		meta.SetResourceCondition(hr, condition, metav1.ConditionFalse, failedReason, msg)
 		r.event(ctx, *hr, revision, events.EventSeverityError, msg)
-		return &ConditionError{Reason: failedReason, Err: errors.New(msg)}
+		return &ConditionError{Reason: failedReason, Err: err}
 	} else {
 		msg := fmt.Sprintf("Helm %s succeeded", action)
 		meta.SetResourceCondition(hr, condition, metav1.ConditionTrue, succeededReason, msg)

internal/runner/log_buffer.go

@@ -0,0 +1,91 @@
+/*
+Copyright 2021 The Flux authors
+
+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 runner
+
+import (
+	"container/ring"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/go-logr/logr"
+	"helm.sh/helm/v3/pkg/action"
+)
+
+const defaultBufferSize = 5
+
+type DebugLog struct {
+	log logr.Logger
+}
+
+func NewDebugLog(log logr.Logger) *DebugLog {
+	return &DebugLog{log: log}
+}
+
+func (l *DebugLog) Log(format string, v ...interface{}) {
+	l.log.V(1).Info(fmt.Sprintf(format, v...))
+}
+
+type LogBuffer struct {
+	mu     sync.Mutex
+	log    action.DebugLog
+	buffer *ring.Ring
+}
+
+func NewLogBuffer(log action.DebugLog, size int) *LogBuffer {
+	if size == 0 {
+		size = defaultBufferSize
+	}
+	return &LogBuffer{
+		log:    log,
+		buffer: ring.New(size),
+	}
+}
+
+func (l *LogBuffer) Log(format string, v ...interface{}) {
+	l.mu.Lock()
+
+	// Filter out duplicate log lines, this happens for example when
+	// Helm is waiting on workloads to become ready.
+	msg := fmt.Sprintf(format, v...)
+	if prev := l.buffer.Prev(); prev.Value != msg {
+		l.buffer.Value = msg
+		l.buffer = l.buffer.Next()
+	}
+
+	l.mu.Unlock()
+	l.log(format, v...)
+}
+
+func (l *LogBuffer) Reset() {
+	l.mu.Lock()
+	l.buffer = ring.New(l.buffer.Len())
+	l.mu.Unlock()
+}
+
+func (l *LogBuffer) String() string {
+	var str string
+	l.mu.Lock()
+	l.buffer.Do(func(s interface{}) {
+		if s == nil {
+			return
+		}
+		str += s.(string) + "\n"
+	})
+	l.mu.Unlock()
+	return strings.TrimSpace(str)
+}

internal/runner/log_buffer_test.go

@@ -0,0 +1,103 @@
+/*
+Copyright 2021 The Flux authors
+
+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 runner
+
+import (
+	"testing"
+
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+func TestLogBuffer_Log(t *testing.T) {
+	tests := []struct {
+		name      string
+		size      int
+		fill      []string
+		wantCount int
+		want      string
+	}{
+		{name: "log", size: 2, fill: []string{"a", "b", "c"}, wantCount: 3, want: "b\nc"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var count int
+			l := NewLogBuffer(func(format string, v ...interface{}) {
+				count++
+				return
+			}, tt.size)
+			for _, v := range tt.fill {
+				l.Log("%s", v)
+			}
+			if count != tt.wantCount {
+				t.Errorf("Inner Log() called %v times, want %v", count, tt.wantCount)
+			}
+			if got := l.String(); got != tt.want {
+				t.Errorf("String() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestLogBuffer_Reset(t *testing.T) {
+	bufferSize := 10
+	l := NewLogBuffer(NewDebugLog(log.NullLogger{}).Log, bufferSize)
+
+	if got := l.buffer.Len(); got != bufferSize {
+		t.Errorf("Len() = %v, want %v", got, bufferSize)
+	}
+
+	for _, v := range []string{"a", "b", "c"} {
+		l.Log("%s", v)
+	}
+
+	if got := l.String(); got == "" {
+		t.Errorf("String() = empty")
+	}
+
+	l.Reset()
+
+	if got := l.buffer.Len(); got != bufferSize {
+		t.Errorf("Len() = %v after Reset(), want %v", got, bufferSize)
+	}
+	if got := l.String(); got != "" {
+		t.Errorf("String() != empty after Reset()")
+	}
+}
+
+func TestLogBuffer_String(t *testing.T) {
+	tests := []struct {
+		name string
+		size int
+		fill []string
+		want string
+	}{
+		{name: "empty buffer", fill: []string{}, want: ""},
+		{name: "filled buffer", size: 2, fill: []string{"a", "b", "c"}, want: "b\nc"},
+		{name: "duplicate buffer items", fill: []string{"b", "b", "b"}, want: "b"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			l := NewLogBuffer(NewDebugLog(log.NullLogger{}).Log, tt.size)
+			for _, v := range tt.fill {
+				l.Log("%s", v)
+			}
+			if got := l.String(); got != tt.want {
+				t.Errorf("String() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

internal/runner/runner.go

@@ -18,7 +18,7 @@ package runner
 
 import (
 	"errors"
-	"fmt"
+	"sync"
 
 	"github.com/go-logr/logr"
 	"helm.sh/helm/v3/pkg/action"
@@ -32,21 +32,39 @@ import (
 	v2 "github.com/fluxcd/helm-controller/api/v2beta1"
 )
 
+type ActionError struct {
+	Err          error
+	CapturedLogs string
+}
+
+func (e ActionError) Error() string {
+	return e.Err.Error()
+}
+
+func (e ActionError) Unwrap() error {
+	return e.Err
+}
+
 // Runner represents a Helm action runner capable of performing Helm
 // operations for a v2beta1.HelmRelease.
 type Runner struct {
-	config *action.Configuration
+	mu        sync.Mutex
+	config    *action.Configuration
+	logBuffer *LogBuffer
 }
 
 // NewRunner constructs a new Runner configured to run Helm actions with the
 // given genericclioptions.RESTClientGetter, and the release and storage
 // namespace configured to the provided values.
 func NewRunner(getter genericclioptions.RESTClientGetter, storageNamespace string, logger logr.Logger) (*Runner, error) {
-	cfg := new(action.Configuration)
+	runner := &Runner{
-	if err := cfg.Init(getter, storageNamespace, "secret", debugLogger(logger)); err != nil {
+		logBuffer: NewLogBuffer(NewDebugLog(logger).Log, 0),
+	}
+	runner.config = new(action.Configuration)
+	if err := runner.config.Init(getter, storageNamespace, "secret", runner.logBuffer.Log); err != nil {
 		return nil, err
 	}
-	return &Runner{config: cfg}, nil
+	return runner, nil
 }
 
 // Create post renderer instances from HelmRelease and combine them into
@@ -67,6 +85,10 @@ func postRenderers(hr v2.HelmRelease) (postrender.PostRenderer, error) {
 
 // Install runs an Helm install action for the given v2beta1.HelmRelease.
 func (r *Runner) Install(hr v2.HelmRelease, chart *chart.Chart, values chartutil.Values) (*release.Release, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	defer r.logBuffer.Reset()
+
 	install := action.NewInstall(r.config)
 	install.ReleaseName = hr.GetReleaseName()
 	install.Namespace = hr.GetReleaseNamespace()
@@ -76,6 +98,7 @@ func (r *Runner) Install(hr v2.HelmRelease, chart *chart.Chart, values chartutil
 	install.DisableOpenAPIValidation = hr.Spec.GetInstall().DisableOpenAPIValidation
 	install.Replace = hr.Spec.GetInstall().Replace
 	install.SkipCRDs = hr.Spec.GetInstall().SkipCRDs
+	install.Devel = true
 	renderer, err := postRenderers(hr)
 	if err != nil {
 		return nil, err
@@ -85,11 +108,16 @@ func (r *Runner) Install(hr v2.HelmRelease, chart *chart.Chart, values chartutil
 		install.CreateNamespace = hr.Spec.GetInstall().CreateNamespace
 	}
 
-	return install.Run(chart, values.AsMap())
+	rel, err := install.Run(chart, values.AsMap())
+	return rel, wrapActionErr(r.logBuffer, err)
 }
 
 // Upgrade runs an Helm upgrade action for the given v2beta1.HelmRelease.
 func (r *Runner) Upgrade(hr v2.HelmRelease, chart *chart.Chart, values chartutil.Values) (*release.Release, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	defer r.logBuffer.Reset()
+
 	upgrade := action.NewUpgrade(r.config)
 	upgrade.Namespace = hr.GetReleaseNamespace()
 	upgrade.ResetValues = !hr.Spec.GetUpgrade().PreserveValues
@@ -100,26 +128,37 @@ func (r *Runner) Upgrade(hr v2.HelmRelease, chart *chart.Chart, values chartutil
 	upgrade.DisableHooks = hr.Spec.GetUpgrade().DisableHooks
 	upgrade.Force = hr.Spec.GetUpgrade().Force
 	upgrade.CleanupOnFail = hr.Spec.GetUpgrade().CleanupOnFail
+	upgrade.Devel = true
 	renderer, err := postRenderers(hr)
 	if err != nil {
 		return nil, err
 	}
 	upgrade.PostRenderer = renderer
 
-	return upgrade.Run(hr.GetReleaseName(), chart, values.AsMap())
+	rel, err := upgrade.Run(hr.GetReleaseName(), chart, values.AsMap())
+	return rel, wrapActionErr(r.logBuffer, err)
 }
 
 // Test runs an Helm test action for the given v2beta1.HelmRelease.
 func (r *Runner) Test(hr v2.HelmRelease) (*release.Release, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	defer r.logBuffer.Reset()
+
 	test := action.NewReleaseTesting(r.config)
 	test.Namespace = hr.GetReleaseNamespace()
 	test.Timeout = hr.Spec.GetTest().GetTimeout(hr.GetTimeout()).Duration
 
-	return test.Run(hr.GetReleaseName())
+	rel, err := test.Run(hr.GetReleaseName())
+	return rel, wrapActionErr(r.logBuffer, err)
 }
 
 // Rollback runs an Helm rollback action for the given v2beta1.HelmRelease.
 func (r *Runner) Rollback(hr v2.HelmRelease) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	defer r.logBuffer.Reset()
+
 	rollback := action.NewRollback(r.config)
 	rollback.Timeout = hr.Spec.GetRollback().GetTimeout(hr.GetTimeout()).Duration
 	rollback.Wait = !hr.Spec.GetRollback().DisableWait
@@ -128,18 +167,23 @@ func (r *Runner) Rollback(hr v2.HelmRelease) error {
 	rollback.Recreate = hr.Spec.GetRollback().Recreate
 	rollback.CleanupOnFail = hr.Spec.GetRollback().CleanupOnFail
 
-	return rollback.Run(hr.GetReleaseName())
+	err := rollback.Run(hr.GetReleaseName())
+	return wrapActionErr(r.logBuffer, err)
 }
 
 // Uninstall runs an Helm uninstall action for the given v2beta1.HelmRelease.
 func (r *Runner) Uninstall(hr v2.HelmRelease) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	defer r.logBuffer.Reset()
+
 	uninstall := action.NewUninstall(r.config)
 	uninstall.Timeout = hr.Spec.GetUninstall().GetTimeout(hr.GetTimeout()).Duration
 	uninstall.DisableHooks = hr.Spec.GetUninstall().DisableHooks
 	uninstall.KeepHistory = hr.Spec.GetUninstall().KeepHistory
 
 	_, err := uninstall.Run(hr.GetReleaseName())
-	return err
+	return wrapActionErr(r.logBuffer, err)
 }
 
 // ObserveLastRelease observes the last revision, if there is one,
@@ -152,8 +196,13 @@ func (r *Runner) ObserveLastRelease(hr v2.HelmRelease) (*release.Release, error)
 	return rel, err
 }
 
-func debugLogger(logger logr.Logger) func(format string, v ...interface{}) {
+func wrapActionErr(log *LogBuffer, err error) error {
-	return func(format string, v ...interface{}) {
+	if err == nil {
-		logger.V(1).Info(fmt.Sprintf(format, v...))
+		return err
 	}
+	err = &ActionError{
+		Err:          err,
+		CapturedLogs: log.String(),
+	}
+	return err
 }