Skip to content

fix(export): implement per-request timeout and fail-fast behavior acr… #443

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

Open
Tamar-Dinavetsky wants to merge 8 commits into
base: main
Choose a base branch
from
Open

fix(export): implement per-request timeout and fail-fast behavior acr… #443

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
35983b8
fix(export): implement per-request timeout and fail-fast behavior acr…
Tamar-Dinavetsky Jun 2, 2026
7d9d41f
test: update unit tests to match new timeout behavior and signatures
Tamar-Dinavetsky Jun 3, 2026
50f9970
fix(export): exit with non-zero code on timeout and add regression tests
Tamar-Dinavetsky Jun 3, 2026
cc72f66
fix(export): move timeout check before manifest writes to prevent par…
Tamar-Dinavetsky Jun 3, 2026
e1f5f0a
fix(context): derive pager context and resolve context leak in CRD loop
Tamar-Dinavetsky Jun 4, 2026
6f9c96a
fix(context): resolve context leaks in discovery pager and iterateItems
Tamar-Dinavetsky Jun 11, 2026
6c5a253
Merge branch 'main' into task/fix-timeout-clean
Tamar-Dinavetsky Jun 18, 2026
d3cfea6
test: update getObjects call site in discover_test to include request…
Tamar-Dinavetsky Jun 18, 2026
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
16 changes: 13 additions & 3 deletions cmd/export/crd.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"fmt"
"strings"
"time"

"github.com/konveyor/crane-lib/apigroups"
"github.com/sirupsen/logrus"
Expand Down Expand Up @@ -85,7 +86,7 @@ func getOperatorManager(obj *unstructured.Unstructured) string {
// API types that appear in resources (deduplicated by plural.group). Built-in API
// groups are skipped. Failed GETs are returned as groupResourceError entries for
// the same failures directory as list errors.
func collectRelatedCRDs(resources []*groupResource, dynamicClient dynamic.Interface, log logrus.FieldLogger, userSkipGroups, userIncludeGroups []string) ([]*groupResource, []*groupResourceError) {
func collectRelatedCRDs(requestTimeout time.Duration, resources []*groupResource, dynamicClient dynamic.Interface, log logrus.FieldLogger, userSkipGroups, userIncludeGroups []string) ([]*groupResource, []*groupResourceError) {
skipSet := normalizeGroupSet(userSkipGroups)
includeSet := normalizeGroupSet(userIncludeGroups)

Expand All @@ -112,7 +113,16 @@ func collectRelatedCRDs(resources []*groupResource, dynamicClient dynamic.Interf
out := make([]*groupResource, 0, len(seen))
var outErrs []*groupResourceError
for crdName := range seen {
obj, err := crdClient.Get(context.Background(), crdName, metav1.GetOptions{})
// Create fresh context with timeout for each CRD Get request
ctx := context.Background()
var cancel context.CancelFunc
if requestTimeout > 0 {
ctx, cancel = context.WithTimeout(ctx, requestTimeout)
}
obj, err := crdClient.Get(ctx, crdName, metav1.GetOptions{})
if cancel != nil {
cancel()
}
if err != nil {
switch {
case apierrors.IsForbidden(err):
Expand All @@ -134,7 +144,7 @@ func collectRelatedCRDs(resources []*groupResource, dynamicClient dynamic.Interf
})
continue
}

if manager := getOperatorManager(obj); manager != "" {
log.Warnf("Skipping CRD %q — managed by %s; install the operator on the target cluster instead", crdName, manager)
continue
Expand Down
54 changes: 42 additions & 12 deletions cmd/export/crd_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/dynamic/fake"
kubetesting "k8s.io/client-go/testing"
)

func TestShouldSkipCRDGroup_DefaultsAndOverrides(t *testing.T) {
Expand Down Expand Up @@ -87,7 +88,7 @@ func TestCollectRelatedCRDs_customResourceOneCRD(t *testing.T) {
client := fake.NewSimpleDynamicClient(scheme, crdUnstructured("widgets.example.com"))
log := testLogger()

got, errs := collectRelatedCRDs([]*groupResource{widgetGroupResource()}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{widgetGroupResource()}, client, log, nil, nil)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand Down Expand Up @@ -120,7 +121,7 @@ func TestCollectRelatedCRDs_builtinGroupNoFetch(t *testing.T) {
Items: []unstructured.Unstructured{{}},
},
}
got, errs := collectRelatedCRDs([]*groupResource{gr}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{gr}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -138,7 +139,7 @@ func TestCollectRelatedCRDs_dedupePluralGroup(t *testing.T) {
w2 := widgetGroupResource()
w2.objects.Items[0].SetName("w2")

got, errs := collectRelatedCRDs([]*groupResource{w1, w2}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{w1, w2}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -155,7 +156,7 @@ func TestCollectRelatedCRDs_skipsSubresourceName(t *testing.T) {
gr := widgetGroupResource()
gr.APIResource.Name = "widgets/status"

got, errs := collectRelatedCRDs([]*groupResource{gr}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{gr}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand Down Expand Up @@ -186,7 +187,7 @@ func TestCollectRelatedCRDs_multipleDistinctCRDs(t *testing.T) {
},
}

got, errs := collectRelatedCRDs([]*groupResource{widgetGroupResource(), gadget}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{widgetGroupResource(), gadget}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -205,7 +206,7 @@ func TestCollectRelatedCRDs_getFailureReturnsGroupResourceError(t *testing.T) {
client := fake.NewSimpleDynamicClient(scheme)
log := testLogger()

got, errs := collectRelatedCRDs([]*groupResource{widgetGroupResource()}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{widgetGroupResource()}, client, log, nil, nil)
if len(got) != 0 {
t.Fatalf("expected no CRD rows, got %d", len(got))
}
Expand Down Expand Up @@ -242,7 +243,7 @@ func TestCollectRelatedCRDs_IncludeOverridesBuiltinGroup(t *testing.T) {
},
}

got, errs := collectRelatedCRDs([]*groupResource{gr}, client, log, nil, []string{"route.openshift.io"})
got, errs := collectRelatedCRDs(0, []*groupResource{gr}, client, log, nil, []string{"route.openshift.io"})
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -260,7 +261,7 @@ func TestCollectRelatedCRDs_skipsOLMManagedCRD(t *testing.T) {
client := fake.NewSimpleDynamicClient(scheme, crd)
log := testLogger()

got, errs := collectRelatedCRDs([]*groupResource{widgetGroupResource()}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{widgetGroupResource()}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -275,7 +276,7 @@ func TestCollectRelatedCRDs_skipsManagedByLabel(t *testing.T) {
client := fake.NewSimpleDynamicClient(scheme, crd)
log := testLogger()

got, errs := collectRelatedCRDs([]*groupResource{widgetGroupResource()}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{widgetGroupResource()}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -292,7 +293,7 @@ func TestCollectRelatedCRDs_skipsOperatorFrameworkAnnotation(t *testing.T) {
client := fake.NewSimpleDynamicClient(scheme, crd)
log := testLogger()

got, errs := collectRelatedCRDs([]*groupResource{widgetGroupResource()}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{widgetGroupResource()}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -307,7 +308,7 @@ func TestCollectRelatedCRDs_skipsOwnerReferenceCRD(t *testing.T) {
client := fake.NewSimpleDynamicClient(scheme, crd)
log := testLogger()

got, errs := collectRelatedCRDs([]*groupResource{widgetGroupResource()}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{widgetGroupResource()}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -322,7 +323,7 @@ func TestCollectRelatedCRDs_exportsUnmanagedCRD(t *testing.T) {
client := fake.NewSimpleDynamicClient(scheme, crd)
log := testLogger()

got, errs := collectRelatedCRDs([]*groupResource{widgetGroupResource()}, client, log, nil, nil)
got, errs := collectRelatedCRDs(0, []*groupResource{widgetGroupResource()}, client, log, nil, nil)
if len(errs) != 0 {
t.Fatalf("unexpected errors: %v", errs)
}
Expand All @@ -331,6 +332,35 @@ func TestCollectRelatedCRDs_exportsUnmanagedCRD(t *testing.T) {
}
}

func TestCollectRelatedCRDs_timeoutError(t *testing.T) {
scheme := runtime.NewScheme()
client := fake.NewSimpleDynamicClient(scheme)

// Simulate timeout error on CRD Get
client.PrependReactor("get", "customresourcedefinitions", func(action kubetesting.Action) (handled bool, ret runtime.Object, err error) {
return true, nil, apierrors.NewTimeoutError("CRD get timeout", 1)
})

log := testLogger()
gr := widgetGroupResource()

// Pass non-zero timeout
got, errs := collectRelatedCRDs(100, []*groupResource{gr}, client, log, nil, nil)

if len(got) != 0 {
t.Fatalf("expected 0 CRDs on timeout, got %d", len(got))
}
if len(errs) != 1 {
t.Fatalf("expected exactly 1 error, got %d: %v", len(errs), errs)
}
if !apierrors.IsTimeout(errs[0].Error) {
t.Fatalf("expected timeout error, got: %v", errs[0].Error)
}
if errs[0].APIResource.Kind != "CustomResourceDefinition" {
t.Fatalf("expected CRD kind in error, got: %s", errs[0].APIResource.Kind)
}
}

func TestGetOperatorManager(t *testing.T) {
tests := []struct {
name string
Expand Down
52 changes: 42 additions & 10 deletions cmd/export/discover.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"os"
"path/filepath"
"strings"
"time"

"github.com/sirupsen/logrus"
apierrors "k8s.io/apimachinery/pkg/api/errors"
Expand Down Expand Up @@ -240,7 +241,7 @@ func discoverPreferredResources(
// resourceToExtract lists objects for each admitted API type in namespace (or cluster-wide
// for allowed cluster-scoped kinds: ClusterRoleBinding, ClusterRole, SCC). It returns resources
// with non-empty lists and a parallel slice of per-type list errors.
func resourceToExtract(namespace string, labelSelector string, dynamicClient dynamic.Interface, lists []*metav1.APIResourceList, log logrus.FieldLogger) ([]*groupResource, []*groupResourceError) {
func resourceToExtract(requestTimeout time.Duration, namespace string, labelSelector string, dynamicClient dynamic.Interface, lists []*metav1.APIResourceList, log logrus.FieldLogger) ([]*groupResource, []*groupResourceError) {
resources := []*groupResource{}
errors := []*groupResourceError{}

Expand Down Expand Up @@ -277,8 +278,13 @@ func resourceToExtract(namespace string, labelSelector string, dynamicClient dyn
APIResource: resource,
}

objs, err := getObjects(g, namespace, labelSelector, dynamicClient, log)
objs, err := getObjects(requestTimeout, g, namespace, labelSelector, dynamicClient, log)
if err != nil {
// Check if error is due to timeout/deadline exceeded - fail fast
if apierrors.IsTimeout(err) || strings.Contains(err.Error(), "context deadline exceeded") {
log.Errorf("request timeout exceeded for groupVersion %s, kind: %s: %v\n", g.APIGroupVersion, g.APIResource.Kind, err)
return nil, []*groupResourceError{{resource, err}}
}
switch {
case apierrors.IsForbidden(err):
log.Debugf("access denied for groupVersion %s, kind: %s (expected for namespace-admin users)\n", g.APIGroupVersion, g.APIResource.Kind)
Expand Down Expand Up @@ -318,30 +324,47 @@ func isAdmittedResource(gv schema.GroupVersion, resource metav1.APIResource) boo

// getObjects lists objects for g using the dynamic client, with paging and optional
// labelSelector. imagestreamtags and imagetags use per-item Get after List.
func getObjects(g *groupResource, namespace string, labelSelector string, d dynamic.Interface, logger logrus.FieldLogger) (*unstructured.UnstructuredList, error) {
func getObjects(requestTimeout time.Duration, g *groupResource, namespace string, labelSelector string, d dynamic.Interface, logger logrus.FieldLogger) (*unstructured.UnstructuredList, error) {
c := d.Resource(schema.GroupVersionResource{
Group: g.APIGroup,
Version: g.APIVersion,
Resource: g.APIResource.Name,
})
p := pager.New(func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
p := pager.New(func(pagerCtx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
// Create a fresh context with timeout for each page request
ctx := pagerCtx
cancel := func() {}

if requestTimeout > 0 {
ctx, cancel = context.WithTimeout(pagerCtx, requestTimeout)
}
defer cancel()

if g.APIResource.Namespaced {
return c.Namespace(namespace).List(context.Background(), opts)
return c.Namespace(namespace).List(ctx, opts)
} else {
return c.List(context.Background(), opts)
return c.List(ctx, opts)
}
})
listOptions := metav1.ListOptions{}
if labelSelector != "" && g.APIResource.Namespaced {
listOptions.LabelSelector = labelSelector
}

list, _, err := p.List(context.TODO(), listOptions)
// Create context for pager.List
ctx := context.Background()
if requestTimeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, requestTimeout)
defer cancel()
}

list, _, err := p.List(ctx, listOptions)
if err != nil {
return nil, err
}
if g.APIResource.Name == "imagestreamtags" || g.APIResource.Name == "imagetags" {
unstructuredList, err := iterateItemsByGet(c, g, list, namespace, logger)
unstructuredList, err := iterateItemsByGet(requestTimeout, c, g, list, namespace, logger)
if err != nil {
return nil, err
}
Expand All @@ -352,7 +375,7 @@ func getObjects(g *groupResource, namespace string, labelSelector string, d dyna

// iterateItemsByGet builds a full UnstructuredList by Get-ing each item name from list
// in namespace (used where List does not return complete objects).
func iterateItemsByGet(c dynamic.NamespaceableResourceInterface, g *groupResource, list runtime.Object, namespace string, logger logrus.FieldLogger) (*unstructured.UnstructuredList, error) {
func iterateItemsByGet(requestTimeout time.Duration, c dynamic.NamespaceableResourceInterface, g *groupResource, list runtime.Object, namespace string, logger logrus.FieldLogger) (*unstructured.UnstructuredList, error) {
if g == nil {
return nil, fmt.Errorf("groupResource cannot be nil")
}
Expand All @@ -364,7 +387,16 @@ func iterateItemsByGet(c dynamic.NamespaceableResourceInterface, g *groupResourc
logger.Errorf("expected unstructured.Unstructured but got %T for groupResource %s and object: %#v\n", object, g.APIResource.Name, object)
return fmt.Errorf("expected *unstructured.Unstructured but got %T", object)
}
obj, err := c.Namespace(namespace).Get(context.TODO(), u.GetName(), metav1.GetOptions{})
// Create fresh context with timeout for each Get request
ctx := context.Background()
cancel := func() {}

if requestTimeout > 0 {
ctx, cancel = context.WithTimeout(ctx, requestTimeout)
}

obj, err := c.Namespace(namespace).Get(ctx, u.GetName(), metav1.GetOptions{})
cancel()
if err != nil {
return err
}
Expand Down
Loading
Loading