Skip to content

fix: retry Helm downloads and extend install timeout #171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
bugkeep wants to merge 1 commit into
base: fix/app-store-prerequisites-review
Choose a base branch
from
Draft

fix: retry Helm downloads and extend install timeout #171

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions conf/app.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@ socks5Proxy = 127.0.0.1:10808
; -- Helm images (optional) -------------------------------------------------
; Set true only when implicit/latest chart images must use IfNotPresent.
helmImplicitLatestPullPolicy = false
; Maximum time Helm waits for an application and its jobs to become ready.
helmInstallTimeout = 20m

; -- Control plane ----------------------------------------------------------
apiserverPort = 6443
Expand Down
8 changes: 8 additions & 0 deletions object/helm_operation.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,14 @@ func StartHelmOperationTaskContext(ctx context.Context, id int64, phase string)
return nil
}

func TouchHelmOperationTaskContext(ctx context.Context, id int64) error {
_, err := ormer.Engine.Context(ctx).ID(id).
Where("status IN (?, ?)", HelmOperationStatusPending, HelmOperationStatusRunning).
Cols("updated_at").
Update(&HelmOperationTask{UpdatedAt: time.Now().UTC()})
return err
}

func UpdateHelmOperationTaskPhase(id int64, phase string) error {
return UpdateHelmOperationTaskPhaseContext(context.Background(), id, phase)
}
Expand Down
56 changes: 35 additions & 21 deletions object/helm_operation_recorder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ const (
helmOperationFinishTimeout = HelmOperationPersistenceTimeout
helmOperationFinishAttempts = 3
helmOperationFinishRetryDelay = 100 * time.Millisecond
helmOperationHeartbeatInterval = time.Minute
)

type HelmOperationRecorder struct {
Expand All @@ -36,32 +37,36 @@ type HelmOperationRecorder struct {
finish sync.Once
finishErr error

persistLogs func(context.Context, int64, []*HelmOperationLog) error
finishTask func(context.Context, int64, bool, string) error
terminalMatches func(context.Context, int64, bool, string) (bool, error)
persistTimeout time.Duration
shutdownTimeout time.Duration
finishTimeout time.Duration
finishRetryDelay time.Duration
persistCtx context.Context
cancelPersist context.CancelFunc
persistLogs func(context.Context, int64, []*HelmOperationLog) error
finishTask func(context.Context, int64, bool, string) error
terminalMatches func(context.Context, int64, bool, string) (bool, error)
touchTask func(context.Context, int64) error
persistTimeout time.Duration
heartbeatInterval time.Duration
shutdownTimeout time.Duration
finishTimeout time.Duration
finishRetryDelay time.Duration
persistCtx context.Context
cancelPersist context.CancelFunc
}

func NewHelmOperationRecorder(taskID int64) *HelmOperationRecorder {
persistCtx, cancelPersist := context.WithCancel(context.Background())
recorder := &HelmOperationRecorder{
taskID: taskID,
queue: make(chan *HelmOperationLog, helmOperationLogBatchSize*2),
done: make(chan struct{}),
persistLogs: addHelmOperationLogsContext,
finishTask: FinishHelmOperationTaskContext,
terminalMatches: HelmOperationTaskHasTerminalOutcomeContext,
persistTimeout: helmOperationLogPersistTimeout,
shutdownTimeout: helmOperationRecorderShutdown,
finishTimeout: helmOperationFinishTimeout,
finishRetryDelay: helmOperationFinishRetryDelay,
persistCtx: persistCtx,
cancelPersist: cancelPersist,
taskID: taskID,
queue: make(chan *HelmOperationLog, helmOperationLogBatchSize*2),
done: make(chan struct{}),
persistLogs: addHelmOperationLogsContext,
finishTask: FinishHelmOperationTaskContext,
terminalMatches: HelmOperationTaskHasTerminalOutcomeContext,
touchTask: TouchHelmOperationTaskContext,
persistTimeout: helmOperationLogPersistTimeout,
heartbeatInterval: helmOperationHeartbeatInterval,
shutdownTimeout: helmOperationRecorderShutdown,
finishTimeout: helmOperationFinishTimeout,
finishRetryDelay: helmOperationFinishRetryDelay,
persistCtx: persistCtx,
cancelPersist: cancelPersist,
}
go recorder.run()
return recorder
Expand Down Expand Up @@ -219,6 +224,8 @@ func (r *HelmOperationRecorder) run() {
}()
ticker := time.NewTicker(helmOperationLogFlushInterval)
defer ticker.Stop()
heartbeatTicker := time.NewTicker(r.heartbeatInterval)
defer heartbeatTicker.Stop()
batch := make([]*HelmOperationLog, 0, helmOperationLogBatchSize)
flush := func() {
if len(batch) == 0 {
Expand Down Expand Up @@ -247,6 +254,13 @@ func (r *HelmOperationRecorder) run() {
}
case <-ticker.C:
flush()
case <-heartbeatTicker.C:
ctx, cancel := context.WithTimeout(r.persistCtx, r.persistTimeout)
err := r.touchTask(ctx, r.taskID)
cancel()
if err != nil && !errors.Is(err, context.Canceled) {
logs.Warning("refresh Helm operation task %d heartbeat: %v", r.taskID, err)
}
}
}
}
148 changes: 76 additions & 72 deletions store/helm.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import (
"bytes"
"context"
"fmt"
"io"
"net/http"
"net/url"
"sort"
Expand Down Expand Up @@ -44,15 +43,13 @@ import (
)

const (
helmOperationTimeout = 5 * time.Minute
helmInstallTimeout = 10 * time.Minute
helmCompatibilityTimeout = 2 * time.Minute
helmChartLoadTimeout = 2 * time.Minute
helmDiagnosticsTimeout = 15 * time.Second
helmInstallOperationTimeout = helmChartLoadTimeout + helmCompatibilityTimeout + helmInstallTimeout + helmDiagnosticsTimeout
helmDiagnosticsMaxEvents = 20
helmDiagnosticsMessageLen = 240
helmDiagnosticsEventLen = 360
helmOperationTimeout = 5 * time.Minute
helmCompatibilityTimeout = 2 * time.Minute
helmChartLoadTimeout = 30 * time.Minute
helmDiagnosticsTimeout = 15 * time.Second
helmDiagnosticsMaxEvents = 20
helmDiagnosticsMessageLen = 240
helmDiagnosticsEventLen = 360
)

// ---------- Types ----------
Expand Down Expand Up @@ -283,35 +280,18 @@ func httpClientWithContext(ctx context.Context, base *http.Client) *http.Client
return &client
}

func helmGet(ctx context.Context, url string) (*http.Response, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return nil, err
}
req.Header.Set("User-Agent", "Helm/3.21.2")
return proxypkg.ProxyHttpClient.Do(req)
}

// ---------- Repo index ----------

func fetchIndexFile(ctx context.Context, repoURL string) (*repo.IndexFile, error) {
indexURL := strings.TrimRight(repoURL, "/") + "/index.yaml"
resp, err := helmGet(ctx, indexURL)
data, err := downloadHelmArtifact(ctx, indexURL)
if err != nil {
return nil, fmt.Errorf(
"fetch index %q: %w",
redactURLForError(indexURL),
sanitizeErrorMessage(err, indexURL, repoURL),
)
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
return nil, fmt.Errorf("index returned HTTP %d", resp.StatusCode)
}
data, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
// Use sigs.k8s.io/yaml (YAML→JSON→struct) so that embedded pointer fields
// like *chart.Metadata inside ChartVersion are properly allocated. Plain
// gopkg.in/yaml.v3 leaves those pointers nil, causing panics in SortEntries.
Expand Down Expand Up @@ -426,7 +406,11 @@ func isOCIChartTag(tag string) bool {
}

func newOCIRegistryClient(ctx context.Context) (*registry.Client, error) {
return registry.NewClient(registry.ClientOptHTTPClient(httpClientWithContext(ctx, proxypkg.ProxyHttpClient)))
client := httpClientWithContext(ctx, proxypkg.ProxyHttpClient)
if client.Timeout <= 0 {
client.Timeout = helmDownloadAttemptTimeout
}
return registry.NewClient(registry.ClientOptHTTPClient(client))
}

// pullOCIChart pulls the chart hosted at repoURL, resolving to the newest published
Expand Down Expand Up @@ -456,7 +440,12 @@ func pullOCIChartFromRegistry(ctx context.Context, repoURL, version string) (*re

if resolvedVersion == "" {
if !strings.Contains(ref, "@") {
tags, err := rc.Tags(ref)
var tags []string
err = retryOCIRegistryOperation(ctx, func() error {
var tagsErr error
tags, tagsErr = rc.Tags(ref)
return tagsErr
})
if err != nil {
return nil, fmt.Errorf(
"list oci tags for %q: %w",
Expand All @@ -481,18 +470,54 @@ func pullOCIChartFromRegistry(ctx context.Context, repoURL, version string) (*re
}
pullRef = fmt.Sprintf("%s:%s", ref, resolvedVersion)
}
if cached, ok, err := cachedOCIChartPull(pullRef); err != nil {
return nil, err
} else if ok {
return cached, nil
}

pull, err := rc.Pull(pullRef, registry.PullOptWithChart(true))
var pull *registry.PullResult
err = retryOCIRegistryOperation(ctx, func() error {
var pullErr error
pull, pullErr = rc.Pull(pullRef, registry.PullOptWithChart(true))
return pullErr
})
if err != nil {
return nil, fmt.Errorf(
"pull oci chart %q: %w",
redactURLForError(repoURL),
sanitizeErrorMessage(err, repoURL, ref, pullRef),
)
}
cacheOCIChartPull(pullRef, pull)
return pull, nil
}

func cachedOCIChartPull(pullRef string) (*registry.PullResult, bool, error) {
data, ok := defaultHelmArtifactCache.get("oci-chart:" + pullRef)
if !ok {
return nil, false, nil
}
loaded, err := loader.LoadArchive(bytes.NewReader(data))
if err != nil {
return nil, true, fmt.Errorf("load cached oci chart %q: %w", redactURLForError(pullRef), err)
}
return &registry.PullResult{
Chart: &registry.DescriptorPullSummaryWithMeta{
DescriptorPullSummary: registry.DescriptorPullSummary{Data: data, Size: int64(len(data))},
Meta: loaded.Metadata,
},
Ref: pullRef,
}, true, nil
}

func cacheOCIChartPull(pullRef string, pull *registry.PullResult) {
if pull == nil || pull.Chart == nil || len(pull.Chart.Data) == 0 {
return
}
defaultHelmArtifactCache.put("oci-chart:"+pullRef, pull.Chart.Data)
}

func latestOCISemverTag(tags []string) string {
type versionedTag struct {
tag string
Expand Down Expand Up @@ -743,7 +768,6 @@ func loadChartWithContext(parent context.Context, chartName, repoURL, version st
}
ctx, cancel := context.WithTimeout(parent, helmChartLoadTimeout)
defer cancel()

if isOCIRepo(repoURL) {
ch, err := loadOCIChart(ctx, repoURL, version)
if err != nil {
Expand Down Expand Up @@ -805,7 +829,7 @@ func loadChartWithContext(parent context.Context, chartName, repoURL, version st
chartURL = strings.TrimRight(repoURL, "/") + "/" + strings.TrimLeft(chartURL, "/")
}

resp, err := helmGet(ctx, chartURL)
data, err := downloadHelmArtifact(ctx, chartURL)
if err != nil {
return nil, fmt.Errorf(
"load chart %q from repo %q version %q: download chart archive %q failed: %w",
Expand All @@ -816,29 +840,6 @@ func loadChartWithContext(parent context.Context, chartName, repoURL, version st
sanitizeErrorMessage(err, chartURL, repoURL),
)
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
return nil, fmt.Errorf(
"load chart %q from repo %q version %q: chart archive %q returned HTTP %d",
chartName,
redactURLForError(repoURL),
entry.Version,
redactURLForError(chartURL),
resp.StatusCode,
)
}

data, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf(
"load chart %q from repo %q version %q: read chart archive %q failed: %w",
chartName,
redactURLForError(repoURL),
entry.Version,
redactURLForError(chartURL),
err,
)
}
ch, err := loader.LoadArchive(bytes.NewReader(data))
if err != nil {
return nil, fmt.Errorf(
Expand Down Expand Up @@ -1267,6 +1268,10 @@ func GetHelmReleases(cfg *rest.Config, namespace string) ([]HelmReleaseSummary,
}

func InstallHelmChart(cfg *rest.Config, releaseName, namespace, chartName, repoURL, version, valuesYAML string) error {
installTimeout, err := configuredHelmInstallTimeout()
if err != nil {
return err
}
actionConfig, err := newHelmConfig(cfg, namespace)
if err != nil {
return err
Expand Down Expand Up @@ -1295,13 +1300,7 @@ func InstallHelmChart(cfg *rest.Config, releaseName, namespace, chartName, repoU
return err
}
install := action.NewInstall(actionConfig)
install.ReleaseName = releaseName
install.Namespace = namespace
install.CreateNamespace = true
install.Wait = true
install.WaitForJobs = true
install.Timeout = helmInstallTimeout
install.PostRenderer = configuredLocalImagePullPolicyPostRenderer()
configureHelmInstall(install, releaseName, namespace, installTimeout)

_, err = install.Run(ch, vals)
if err != nil {
Expand Down Expand Up @@ -1336,8 +1335,6 @@ func InstallHelmChartStream(ctx context.Context, lifecycle HelmInstallLifecycle,
if streamCtx == nil {
streamCtx = context.Background()
}
installCtx, cancelInstall := context.WithTimeout(context.WithoutCancel(streamCtx), helmInstallOperationTimeout)
defer cancelInstall()
send := func(line string) bool {
if err := lifecycle.RecordLog(line); err != nil {
logrus.Warnf("failed to persist Helm operation log: %v", err)
Expand Down Expand Up @@ -1367,6 +1364,19 @@ func InstallHelmChartStream(ctx context.Context, lifecycle HelmInstallLifecycle,
}
return
}
installTimeout, err := configuredHelmInstallTimeout()
if err != nil {
send("ERROR: " + err.Error())
if finishErr := lifecycle.Finish(err); finishErr != nil {
logrus.Errorf("failed to finish Helm operation after configuration error: %v", finishErr)
}
return
}
installCtx, cancelInstall := context.WithTimeout(
context.WithoutCancel(streamCtx),
helmChartLoadTimeout+helmCompatibilityTimeout+installTimeout+helmDiagnosticsTimeout,
)
defer cancelInstall()
helmChart, err := loadChartWithContext(installCtx, chartName, repoURL, version)
if err != nil {
send("ERROR: " + err.Error())
Expand Down Expand Up @@ -1411,13 +1421,7 @@ func InstallHelmChartStream(ctx context.Context, lifecycle HelmInstallLifecycle,
return
}
install := action.NewInstall(actionConfig)
install.ReleaseName = releaseName
install.Namespace = namespace
install.CreateNamespace = true
install.Wait = true
install.WaitForJobs = true
install.Timeout = helmInstallTimeout
install.PostRenderer = configuredLocalImagePullPolicyPostRenderer()
configureHelmInstall(install, releaseName, namespace, installTimeout)
if _, err = install.RunWithContext(installCtx, helmChart, vals); err != nil {
for _, line := range helmReleaseDiagnostics(installCtx, cfg, releaseName, namespace) {
send(line)
Expand Down
Loading
Loading