jayjay
/
k9s
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add exponential backoff

mine
derailed 2020-10-27 20:03:09 -06:00
parent 087e6643f9
commit dd9bddf78f
8 changed files with 43 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/dao/registry.go
View File

@ -135,7 +135,7 @@ func (m *Meta) LoadResources(f Factory) error {
return nil
}
// BOZO!! Need contermeasure for direct commands!
// BOZO!! Need countermeasures for direct commands!
func loadNonResource(m ResourceMetas) {
loadK9s(m)
loadRBAC(m)

9
internal/model/describe.go
View File

@ -26,7 +26,7 @@ type ResourceViewer interface {
Filter(string)
ClearFilter()
Peek() []string
Watch(context.Context)
Watch(context.Context) error
AddListener(ResourceViewerListener)
RemoveListener(ResourceViewerListener)
}
@ -117,9 +117,12 @@ func (d *Describe) Peek() []string {
}
// Watch watches for describe data changes.
func (d *Describe) Watch(ctx context.Context) {
d.refresh(ctx)
func (d *Describe) Watch(ctx context.Context) error {
if err := d.refresh(ctx); err != nil {
return err
}
go d.updater(ctx)
return nil
}
func (d *Describe) updater(ctx context.Context) {

64
internal/model/table.go
View File

@ -7,6 +7,7 @@ import (
"sync/atomic"
"time"
backoff "github.com/cenkalti/backoff/v4"
"github.com/derailed/k9s/internal"
"github.com/derailed/k9s/internal/client"
"github.com/derailed/k9s/internal/dao"
@ -84,14 +85,18 @@ func (t *Table) RemoveListener(l TableListener) {
}
// Watch initiates model updates.
func (t *Table) Watch(ctx context.Context) {
t.refresh(ctx)
func (t *Table) Watch(ctx context.Context) error {
if err := t.refresh(ctx); err != nil {
return err
}
go t.updater(ctx)
return nil
}
// Refresh updates the table content.
func (t *Table) Refresh(ctx context.Context) {
t.refresh(ctx)
func (t *Table) Refresh(ctx context.Context) error {
return t.refresh(ctx)
}
// Get returns a resource instance if found, else an error.
@ -119,36 +124,6 @@ func (t *Table) Delete(ctx context.Context, path string, cascade, force bool) er
return nuker.Delete(path, cascade, force)
}
// Describe describes a given resource.
func (t *Table) Describe(ctx context.Context, path string) (string, error) {
meta, err := getMeta(ctx, t.gvr)
if err != nil {
return "", err
}
desc, ok := meta.DAO.(dao.Describer)
if !ok {
return "", fmt.Errorf("no describer for %q", meta.DAO.GVR())
}
return desc.Describe(path)
}
// ToYAML returns a resource yaml.
func (t *Table) ToYAML(ctx context.Context, path string) (string, error) {
meta, err := getMeta(ctx, t.gvr)
if err != nil {
return "", err
}
desc, ok := meta.DAO.(dao.Describer)
if !ok {
return "", fmt.Errorf("no describer for %q", meta.DAO.GVR())
}
return desc.ToYAML(path, false)
}
// GetNamespace returns the model namespace.
func (t *Table) GetNamespace() string {
return t.namespace
@ -191,6 +166,8 @@ func (t *Table) Peek() render.TableData {
func (t *Table) updater(ctx context.Context) {
defer log.Debug().Msgf("TABLE-MODEL canceled -- %q", t.gvr)
bf := backoff.NewExponentialBackOff()
bf.InitialInterval, bf.MaxElapsedTime = initRefreshRate, maxRetryInterval
rate := initRefreshRate
for {
select {
@ -198,24 +175,31 @@ func (t *Table) updater(ctx context.Context) {
return
case <-time.After(rate):
rate = t.refreshRate
t.refresh(ctx)
err := backoff.Retry(func() error {
return t.refresh(ctx)
}, backoff.WithContext(bf, ctx))
if err != nil {
log.Error().Err(err).Msgf("Retry failed")
t.fireTableLoadFailed(err)
return
}
}
}
}
func (t *Table) refresh(ctx context.Context) {
func (t *Table) refresh(ctx context.Context) error {
if !atomic.CompareAndSwapInt32(&t.inUpdate, 0, 1) {
log.Debug().Msgf("Dropping update...")
return
return nil
}
defer atomic.StoreInt32(&t.inUpdate, 0)
if err := t.reconcile(ctx); err != nil {
log.Error().Err(err).Msgf("reconcile failed %q::%q", t.gvr, t.instance)
t.fireTableLoadFailed(err)
return
return err
}
t.fireTableChanged(t.Peek())
return nil
}
func (t *Table) list(ctx context.Context, a dao.Accessor) ([]runtime.Object, error) {

9
internal/model/yaml.go
View File

@ -108,12 +108,13 @@ func (y *YAML) Peek() []string {
}
// Watch watches for YAML changes.
func (y *YAML) Watch(ctx context.Context) {
func (y *YAML) Watch(ctx context.Context) error {
if err := y.refresh(ctx); err != nil {
log.Error().Err(err).Msgf("YAML Refresh failed")
return
return err
}
go y.updater(ctx)
return nil
}
func (y *YAML) updater(ctx context.Context) {
@ -148,8 +149,6 @@ func (y *YAML) refresh(ctx context.Context) error {
defer atomic.StoreInt32(&y.inUpdate, 0)
if err := y.reconcile(ctx); err != nil {
log.Error().Err(err).Msgf("reconcile failed %q", y.gvr)
y.fireResourceFailed(err)
return err
}

4
internal/ui/table_test.go
View File

@ -66,14 +66,14 @@ func (t *mockModel) SetLabelFilter(string) {}
func (t *mockModel) Empty() bool { return false }
func (t *mockModel) HasMetrics() bool { return true }
func (t *mockModel) Peek() render.TableData { return makeTableData() }
func (t *mockModel) Refresh(context.Context) {}
func (t *mockModel) Refresh(context.Context) error { return nil }
func (t *mockModel) ClusterWide() bool { return false }
func (t *mockModel) GetNamespace() string { return "blee" }
func (t *mockModel) SetNamespace(string) {}
func (t *mockModel) ToggleToast() {}
func (t *mockModel) AddListener(model.TableListener) {}
func (t *mockModel) RemoveListener(model.TableListener) {}
func (t *mockModel) Watch(context.Context) {}
func (t *mockModel) Watch(context.Context) error { return nil }
func (t *mockModel) Get(ctx context.Context, path string) (runtime.Object, error) {
return nil, nil
}

10
internal/ui/types.go
View File

@ -39,12 +39,6 @@ type Namespaceable interface {
type Lister interface {
// Get returns a resource instance.
Get(ctx context.Context, path string) (runtime.Object, error)
// ToYAML returns a resource yaml representation.
ToYAML(ctx context.Context, path string) (string, error)
// Describes describes a given resource.
Describe(ctx context.Context, path string) (string, error)
}
// Tabular represents a tabular model.
@ -65,10 +59,10 @@ type Tabular interface {
Peek() render.TableData
// Watch watches a given resource for changes.
Watch(context.Context)
Watch(context.Context) error
// Refresh forces a new refresh.
Refresh(context.Context)
Refresh(context.Context) error
// SetRefreshRate sets the model watch loop rate.
SetRefreshRate(time.Duration)

4
internal/view/alias_test.go
View File

@ -116,8 +116,8 @@ func (t *mockModel) SetNamespace(string) {}
func (t *mockModel) ToggleToast() {}
func (t *mockModel) AddListener(model.TableListener) {}
func (t *mockModel) RemoveListener(model.TableListener) {}
func (t *mockModel) Watch(context.Context) {}
func (t *mockModel) Refresh(context.Context) {}
func (t *mockModel) Watch(context.Context) error { return nil }
func (t *mockModel) Refresh(context.Context) error { return nil }
func (t *mockModel) Get(context.Context, string) (runtime.Object, error) {
return nil, nil

4
internal/view/table_int_test.go
View File

@ -97,14 +97,14 @@ func (t *mockTableModel) SetLabelFilter(string) {}
func (t *mockTableModel) Empty() bool { return false }
func (t *mockTableModel) HasMetrics() bool { return true }
func (t *mockTableModel) Peek() render.TableData { return makeTableData() }
func (t *mockTableModel) Refresh(context.Context) {}
func (t *mockTableModel) Refresh(context.Context) error { return nil }
func (t *mockTableModel) ClusterWide() bool { return false }
func (t *mockTableModel) GetNamespace() string { return "blee" }
func (t *mockTableModel) SetNamespace(string) {}
func (t *mockTableModel) ToggleToast() {}
func (t *mockTableModel) AddListener(model.TableListener) {}
func (t *mockTableModel) RemoveListener(model.TableListener) {}
func (t *mockTableModel) Watch(context.Context) {}
func (t *mockTableModel) Watch(context.Context) error { return nil }
func (t *mockTableModel) Get(context.Context, string) (runtime.Object, error) {
return nil, nil
}