#5195

Committed 18 Mar 2025 12:53PM UTC coverage: 59.347%. First build

Build # #5195

Build Type

Pull #3681

travis-ci

Committed by

akalenyu

Commit Message

Avoid waiting out entire tcp timeout in progress call

Pretty sure we lose a worker thread whenever the GET call to /metrics
gets issued against a completed pod. Reducing the timeout to 2 seconds
should minimize that window.

```
{"level":"error","ts":"2025-03-17T20:40:34Z","logger":"controller.import-populator","msg":"Failed to update import progress for pvc <NAME/NS>","error":"Get \"https://<IP>:8443/metrics\": dial tcp <IP>:8443: i/o timeout"
```

Signed-off-by: Alex Kalenyuk <akalenyu@redhat.com>

Pull Request Pull Request #3681: Avoid waiting out entire tcp timeout in progress call

Run Details

3 of 14 new or added lines in 5 files covered. (21.43%)

16800 of 28308 relevant lines covered (59.35%)

0.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

38.4

/pkg/controller/clone/host-clone.go

package clone

import (
        "context"
        "fmt"
        "net/http"
        "strconv"
        "time"

        "github.com/go-logr/logr"

        corev1 "k8s.io/api/core/v1"
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/client-go/tools/record"

        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/reconcile"

        cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
        cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
        metrics "kubevirt.io/containerized-data-importer/pkg/monitoring/metrics/cdi-cloner"
)

// HostClonePhaseName is the name of the host clone phase
const HostClonePhaseName = "HostClone"

// HostClonePhase creates and monitors a dumb clone operation
type HostClonePhase struct {
        Owner             client.Object
        Namespace         string
        SourceName        string
        DesiredClaim      *corev1.PersistentVolumeClaim
        ImmediateBind     bool
        OwnershipLabel    string
        Preallocation     bool
        PriorityClassName string
        Client            client.Client
        Log               logr.Logger
        Recorder          record.EventRecorder
}

var _ Phase = &HostClonePhase{}

var _ StatusReporter = &HostClonePhase{}

var httpClient *http.Client

func init() {
        httpClient = cc.BuildHTTPClient(httpClient)
}

// Name returns the name of the phase
func (p *HostClonePhase) Name() string {
        return HostClonePhaseName
}

// Status returns the phase status
func (p *HostClonePhase) Status(ctx context.Context) (*PhaseStatus, error) {
        result := &PhaseStatus{}
        pvc := &corev1.PersistentVolumeClaim{}
        exists, err := getResource(ctx, p.Client, p.Namespace, p.DesiredClaim.Name, pvc)
        if err != nil {
                return nil, err
        }

        if !exists {
                return result, nil
        }

        result.Annotations = pvc.Annotations

        podName := pvc.Annotations[cc.AnnCloneSourcePod]
        if podName == "" {
                return result, nil
        }

        args := &progressFromClaimArgs{
                Client:       p.Client,
                HTTPClient:   httpClient,
                Claim:        pvc,
                PodNamespace: p.Namespace,
                PodName:      podName,
                OwnerUID:     string(p.Owner.GetUID()),
        }

        progress, err := progressFromClaim(ctx, args)
        if err != nil {
                return nil, err
        }

        result.Progress = progress

        return result, nil
}

// progressFromClaimArgs are the args for ProgressFromClaim
type progressFromClaimArgs struct {
        Client       client.Client
        HTTPClient   *http.Client
        Claim        *corev1.PersistentVolumeClaim
        OwnerUID     string
        PodNamespace string
        PodName      string
}

// progressFromClaim returns the progres
func progressFromClaim(ctx context.Context, args *progressFromClaimArgs) (string, error) {
        // Just set 100.0% if pod is succeeded
        if args.Claim.Annotations[cc.AnnPodPhase] == string(corev1.PodSucceeded) {
                return cc.ProgressDone, nil
        }

        pod := &corev1.Pod{
                ObjectMeta: metav1.ObjectMeta{
                        Namespace: args.PodNamespace,
                        Name:      args.PodName,
                },
        }
        if err := args.Client.Get(ctx, client.ObjectKeyFromObject(pod), pod); err != nil {
                if k8serrors.IsNotFound(err) {
                        return "", nil
                }
                return "", err
        }

        // This will only work when the clone source pod is running
        if pod.Status.Phase != corev1.PodRunning {
                return "", nil
        }
        url, err := cc.GetMetricsURL(pod)
        if err != nil {
                return "", err
        }
        if url == "" {
                return "", nil
        }

        // We fetch the clone progress from the clone source pod metrics
        progressReport, err := cc.GetProgressReportFromURL(ctx, url, args.HTTPClient, metrics.CloneProgressMetricName, args.OwnerUID)
        if err != nil {
                return "", err
        }
        if progressReport != "" {
                if f, err := strconv.ParseFloat(progressReport, 64); err == nil {
                        return fmt.Sprintf("%.2f%%", f), nil
                }
        }

        return "", nil
}

// Reconcile creates the desired pvc and waits for the operation to complete
func (p *HostClonePhase) Reconcile(ctx context.Context) (*reconcile.Result, error) {
        actualClaim := &corev1.PersistentVolumeClaim{}
        exists, err := getResource(ctx, p.Client, p.Namespace, p.DesiredClaim.Name, actualClaim)
        if err != nil {
                return nil, err
        }

        if !exists {
                actualClaim, err = p.createClaim(ctx)
                if err != nil {
                        return nil, err
                }
        }

        if !p.hostCloneComplete(actualClaim) {
                // requeue to update status
                return &reconcile.Result{RequeueAfter: 3 * time.Second}, nil
        }

        return nil, nil
}

func (p *HostClonePhase) createClaim(ctx context.Context) (*corev1.PersistentVolumeClaim, error) {
        claim := p.DesiredClaim.DeepCopy()

        claim.Namespace = p.Namespace
        cc.AddAnnotation(claim, cc.AnnPreallocationRequested, fmt.Sprintf("%t", p.Preallocation))
        cc.AddAnnotation(claim, cc.AnnOwnerUID, string(p.Owner.GetUID()))
        cc.AddAnnotation(claim, cc.AnnPodRestarts, "0")
        cc.AddAnnotation(claim, cc.AnnCloneRequest, fmt.Sprintf("%s/%s", p.Namespace, p.SourceName))
        cc.AddAnnotation(claim, cc.AnnPopulatorKind, cdiv1.VolumeCloneSourceRef)
        cc.AddAnnotation(claim, cc.AnnEventSourceKind, p.Owner.GetObjectKind().GroupVersionKind().Kind)
        cc.AddAnnotation(claim, cc.AnnEventSource, fmt.Sprintf("%s/%s", p.Owner.GetNamespace(), p.Owner.GetName()))
        if p.OwnershipLabel != "" {
                AddOwnershipLabel(p.OwnershipLabel, claim, p.Owner)
        }
        if p.ImmediateBind {
                cc.AddAnnotation(claim, cc.AnnImmediateBinding, "")
        }
        if p.PriorityClassName != "" {
                cc.AddAnnotation(claim, cc.AnnPriorityClassName, p.PriorityClassName)
        }
        cc.AddLabel(claim, cc.LabelExcludeFromVeleroBackup, "true")

        if err := p.Client.Create(ctx, claim); err != nil {
                checkQuotaExceeded(p.Recorder, p.Owner, err)
                return nil, err
        }

        return claim, nil
}

func (p *HostClonePhase) hostCloneComplete(pvc *corev1.PersistentVolumeClaim) bool {
        // this is awfully lame
        // both the upload controller and clone controller update the PVC status to succeeded
        // but only the clone controller will set the preallocation annotation
        // so we have to wait for that
        if p.Preallocation && pvc.Annotations[cc.AnnPreallocationApplied] != "true" {
                return false
        }
        return pvc.Annotations[cc.AnnPodPhase] == string(cdiv1.Succeeded)
}

1	package clone
2
3	import (
4	"context"
5	"fmt"
6	"net/http"
7	"strconv"
8	"time"
9
10	"github.com/go-logr/logr"
11
12	corev1 "k8s.io/api/core/v1"
13	k8serrors "k8s.io/apimachinery/pkg/api/errors"
14	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
15	"k8s.io/client-go/tools/record"
16
17	"sigs.k8s.io/controller-runtime/pkg/client"
18	"sigs.k8s.io/controller-runtime/pkg/reconcile"
19
20	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
21	cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
22	metrics "kubevirt.io/containerized-data-importer/pkg/monitoring/metrics/cdi-cloner"
23	)
24
25	// HostClonePhaseName is the name of the host clone phase
26	const HostClonePhaseName = "HostClone"
27
28	// HostClonePhase creates and monitors a dumb clone operation
29	type HostClonePhase struct {
30	Owner client.Object
31	Namespace string
32	SourceName string
33	DesiredClaim *corev1.PersistentVolumeClaim
34	ImmediateBind bool
35	OwnershipLabel string
36	Preallocation bool
37	PriorityClassName string
38	Client client.Client
39	Log logr.Logger
40	Recorder record.EventRecorder
41	}
42
43	var _ Phase = &HostClonePhase{}
44
45	var _ StatusReporter = &HostClonePhase{}
46
47	var httpClient *http.Client
48
49	func init() {	1✔
50	httpClient = cc.BuildHTTPClient(httpClient)	1✔
51	}	1✔
52
53	// Name returns the name of the phase
54	func (p *HostClonePhase) Name() string {	×
55	return HostClonePhaseName	×
56	}	×
57
58	// Status returns the phase status
59	func (p HostClonePhase) Status(ctx context.Context) (PhaseStatus, error) {	×
60	result := &PhaseStatus{}	×
61	pvc := &corev1.PersistentVolumeClaim{}	×
62	exists, err := getResource(ctx, p.Client, p.Namespace, p.DesiredClaim.Name, pvc)	×
63	if err != nil {	×
64	return nil, err	×
65	}	×
66
67	if !exists {	×
68	return result, nil	×
69	}	×
70
71	result.Annotations = pvc.Annotations	×
72		×
73	podName := pvc.Annotations[cc.AnnCloneSourcePod]	×
74	if podName == "" {	×
75	return result, nil	×
76	}	×
77
78	args := &progressFromClaimArgs{	×
79	Client: p.Client,	×
80	HTTPClient: httpClient,	×
81	Claim: pvc,	×
82	PodNamespace: p.Namespace,	×
83	PodName: podName,	×
84	OwnerUID: string(p.Owner.GetUID()),	×
85	}	×
86		×
87	progress, err := progressFromClaim(ctx, args)	×
88	if err != nil {	×
89	return nil, err	×
90	}	×
91
92	result.Progress = progress	×
93		×
94	return result, nil	×
95	}
96
97	// progressFromClaimArgs are the args for ProgressFromClaim
98	type progressFromClaimArgs struct {
99	Client client.Client
100	HTTPClient *http.Client
101	Claim *corev1.PersistentVolumeClaim
102	OwnerUID string
103	PodNamespace string
104	PodName string
105	}
106
107	// progressFromClaim returns the progres
108	func progressFromClaim(ctx context.Context, args *progressFromClaimArgs) (string, error) {	×
109	// Just set 100.0% if pod is succeeded	×
110	if args.Claim.Annotations[cc.AnnPodPhase] == string(corev1.PodSucceeded) {	×
111	return cc.ProgressDone, nil	×
112	}	×
113
114	pod := &corev1.Pod{	×
115	ObjectMeta: metav1.ObjectMeta{	×
116	Namespace: args.PodNamespace,	×
117	Name: args.PodName,	×
118	},	×
119	}	×
120	if err := args.Client.Get(ctx, client.ObjectKeyFromObject(pod), pod); err != nil {	×
121	if k8serrors.IsNotFound(err) {	×
122	return "", nil	×
123	}	×
124	return "", err	×
125	}
126
127	// This will only work when the clone source pod is running
128	if pod.Status.Phase != corev1.PodRunning {	×
129	return "", nil	×
130	}	×
131	url, err := cc.GetMetricsURL(pod)	×
132	if err != nil {	×
133	return "", err	×
134	}	×
135	if url == "" {	×
136	return "", nil	×
137	}	×
138
139	// We fetch the clone progress from the clone source pod metrics
NEW 140	progressReport, err := cc.GetProgressReportFromURL(ctx, url, args.HTTPClient, metrics.CloneProgressMetricName, args.OwnerUID)	×
141	if err != nil {	×
142	return "", err	×
143	}	×
144	if progressReport != "" {	×
145	if f, err := strconv.ParseFloat(progressReport, 64); err == nil {	×
146	return fmt.Sprintf("%.2f%%", f), nil	×
147	}	×
148	}
149
150	return "", nil	×
151	}
152
153	// Reconcile creates the desired pvc and waits for the operation to complete
154	func (p HostClonePhase) Reconcile(ctx context.Context) (reconcile.Result, error) {	1✔
155	actualClaim := &corev1.PersistentVolumeClaim{}	1✔
156	exists, err := getResource(ctx, p.Client, p.Namespace, p.DesiredClaim.Name, actualClaim)	1✔
157	if err != nil {	1✔
158	return nil, err	×
159	}	×
160
161	if !exists {	2✔
162	actualClaim, err = p.createClaim(ctx)	1✔
163	if err != nil {	1✔
164	return nil, err	×
165	}	×
166	}
167
168	if !p.hostCloneComplete(actualClaim) {	2✔
169	// requeue to update status	1✔
170	return &reconcile.Result{RequeueAfter: 3 * time.Second}, nil	1✔
171	}	1✔
172
173	return nil, nil	1✔
174	}
175
176	func (p HostClonePhase) createClaim(ctx context.Context) (corev1.PersistentVolumeClaim, error) {	1✔
177	claim := p.DesiredClaim.DeepCopy()	1✔
178		1✔
179	claim.Namespace = p.Namespace	1✔
180	cc.AddAnnotation(claim, cc.AnnPreallocationRequested, fmt.Sprintf("%t", p.Preallocation))	1✔
181	cc.AddAnnotation(claim, cc.AnnOwnerUID, string(p.Owner.GetUID()))	1✔
182	cc.AddAnnotation(claim, cc.AnnPodRestarts, "0")	1✔
183	cc.AddAnnotation(claim, cc.AnnCloneRequest, fmt.Sprintf("%s/%s", p.Namespace, p.SourceName))	1✔
184	cc.AddAnnotation(claim, cc.AnnPopulatorKind, cdiv1.VolumeCloneSourceRef)	1✔
185	cc.AddAnnotation(claim, cc.AnnEventSourceKind, p.Owner.GetObjectKind().GroupVersionKind().Kind)	1✔
186	cc.AddAnnotation(claim, cc.AnnEventSource, fmt.Sprintf("%s/%s", p.Owner.GetNamespace(), p.Owner.GetName()))	1✔
187	if p.OwnershipLabel != "" {	2✔
188	AddOwnershipLabel(p.OwnershipLabel, claim, p.Owner)	1✔
189	}	1✔
190	if p.ImmediateBind {	2✔
191	cc.AddAnnotation(claim, cc.AnnImmediateBinding, "")	1✔
192	}	1✔
193	if p.PriorityClassName != "" {	2✔
194	cc.AddAnnotation(claim, cc.AnnPriorityClassName, p.PriorityClassName)	1✔
195	}	1✔
196	cc.AddLabel(claim, cc.LabelExcludeFromVeleroBackup, "true")	1✔
197		1✔
198	if err := p.Client.Create(ctx, claim); err != nil {	1✔
199	checkQuotaExceeded(p.Recorder, p.Owner, err)	×
200	return nil, err	×
201	}	×
202
203	return claim, nil	1✔
204	}
205
206	func (p HostClonePhase) hostCloneComplete(pvc corev1.PersistentVolumeClaim) bool {	1✔
207	// this is awfully lame	1✔
208	// both the upload controller and clone controller update the PVC status to succeeded	1✔
209	// but only the clone controller will set the preallocation annotation	1✔
210	// so we have to wait for that	1✔
211	if p.Preallocation && pvc.Annotations[cc.AnnPreallocationApplied] != "true" {	2✔
212	return false	1✔
213	}	1✔
214	return pvc.Annotations[cc.AnnPodPhase] == string(cdiv1.Succeeded)	1✔
215	}

kubevirt / containerized-data-importer / #5195

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous