forked from oVirt/go-ovirt-client
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathclient_disk_downloadimage.go
282 lines (255 loc) · 7.93 KB
/
client_disk_downloadimage.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
package ovirtclient
import (
"context"
"fmt"
"io"
"net/http"
"strconv"
"sync"
ovirtsdk4 "github.com/ovirt/go-ovirt"
)
// Deprecated: use StartDownloadDisk instead.
func (o *oVirtClient) StartImageDownload(diskID string, format ImageFormat, retries ...RetryStrategy) (ImageDownload, error) {
o.logger.Debugf("Using StartImageDownload is deprecated, please use StartDownloadDisk instead.")
return o.StartDownloadDisk(diskID, format, retries...)
}
func (o *oVirtClient) StartDownloadDisk(diskID string, format ImageFormat, retries ...RetryStrategy) (ImageDownload, error) {
retries = defaultRetries(retries, defaultLongTimeouts())
o.logger.Infof("Starting disk %s image download...", diskID)
disk, err := o.GetDisk(diskID)
if err != nil {
return nil, wrap(err, EUnidentified, "failed to fetch disk for image download")
}
realCtx, cancel := context.WithCancel(context.Background())
dl := &imageDownload{
disk: disk,
lock: &sync.Mutex{},
bytesRead: 0,
size: disk.TotalSize(),
lastError: nil,
ctx: realCtx,
cancel: cancel,
conn: o.conn,
done: make(chan struct{}),
reader: nil,
httpClient: o.httpClient,
createReq: nil,
transfer: nil,
cli: o,
logger: o.logger,
retries: retries,
format: format,
}
go dl.poll()
return dl, nil
}
// Deprecated: use DownloadDisk instead.
func (o *oVirtClient) DownloadImage(diskID string, format ImageFormat, retries ...RetryStrategy) (
ImageDownloadReader,
error,
) {
o.logger.Debugf("Using DownloadImage is deprecated, please use DownloadDisk instead.")
return o.DownloadDisk(diskID, format, retries...)
}
func (o *oVirtClient) DownloadDisk(
diskID string,
format ImageFormat,
retries ...RetryStrategy,
) (ImageDownloadReader, error) {
download, err := o.StartDownloadDisk(diskID, format, retries...)
if err != nil {
return nil, err
}
<-download.Initialized()
if err := download.Err(); err != nil {
_ = download.Close()
return nil, err
}
return download, nil
}
type imageDownload struct {
lock *sync.Mutex
bytesRead uint64
size uint64
lastError error
ctx context.Context
cancel context.CancelFunc
conn *ovirtsdk4.Connection
done chan struct{}
reader io.ReadCloser
httpClient http.Client
createReq *ovirtsdk4.ImageTransfersServiceAddRequest
transfer imageTransfer
cli *oVirtClient
logger Logger
retries []RetryStrategy
format ImageFormat
disk Disk
}
// poll polls the oVirt API for the status of the transfer and initializes the HTTP request to
// download the image. Once the HTTP transfer is available control is handed back and the
// imageDownload becomes a reader to read the data from. When Close is called on imageDownload
// the transfer is finalized.
func (i *imageDownload) poll() {
defer close(i.done)
i.transfer = newImageTransfer(
i.cli,
i.logger,
i.disk.ID(),
"",
i.retries,
ovirtsdk4.IMAGETRANSFERDIRECTION_DOWNLOAD,
ovirtsdk4.DiskFormat(i.format),
i.updateDisk,
)
transferURL := ""
var err error
if transferURL, err = i.transfer.initialize(); err != nil {
i.lastError = i.transfer.finalize(err)
return
}
var httpResponse *http.Response
httpResponse, err = i.transferImage(transferURL) //nolint:bodyclose
if err != nil {
i.lastError = i.transfer.finalize(err)
return
}
i.reader = httpResponse.Body
}
// updateDisk is a helper function that updates the internally-stored disk object when the transfer updates it.
func (i *imageDownload) updateDisk(disk Disk) {
i.disk = disk
}
// transferImage will retry a HTTP GET request to download the image and return the HTTP response if successful.
// This call will also set the exact download size in i.size. This function will retry until a valid URL is obtained
// or retries are exhausted.
func (i *imageDownload) transferImage(transferURL string) (httpResponse *http.Response, err error) {
return httpResponse, retry(
fmt.Sprintf("transferring image from %s", transferURL),
i.logger,
i.retries,
func() error {
response, err := i.attemptTransferImage(transferURL) //nolint:bodyclose
httpResponse = response
return err
},
)
}
// attemptTransferImage will create a single attempt to download an image from the specified transfer URL.
func (i *imageDownload) attemptTransferImage(transferURL string) (*http.Response, error) {
req, err := http.NewRequest("GET", transferURL, nil)
if err != nil {
return nil, wrap(err, EBug, "failed to create HTTP request to %s", transferURL)
}
httpResponse, err := i.httpClient.Do(req)
if err != nil {
return httpResponse, wrap(
err,
EConnection,
"HTTP request to image transfer URL %s failed",
transferURL,
)
}
if err := i.transfer.checkStatusCode(httpResponse.StatusCode); err != nil {
_ = httpResponse.Body.Close()
return nil, wrap(
err,
EUnidentified,
"failed to download image from disk %s",
i.disk.ID(),
)
}
if err := i.extractDownloadSize(httpResponse); err != nil {
_ = httpResponse.Body.Close()
return nil, wrap(
err,
EUnidentified,
"failed to determine download size for disk %s",
i.disk.ID(),
)
}
return httpResponse, nil
}
// extractDownloadSize extracts the size of the image download from the content-length header of the
// HTTP response if needed. This is a fallback mechanism for the case when the engine doesn't return
// the correct image size.
func (i *imageDownload) extractDownloadSize(httpResponse *http.Response) error {
if i.size != 0 {
return nil
}
header := httpResponse.Header
if header.Get("content-encoding") != "" {
return newError(
EBug,
"the oVirt engine API did not return an image download size and the ImageIO response contained a non-plaintext encoding, so the download size cannot be determined",
)
}
if contentLengthString := header.Get("content-length"); contentLengthString != "" {
contentLength, err := strconv.ParseUint(contentLengthString, 10, 64)
if err != nil {
return wrap(
err,
EBug,
"the oVirt engine API did not return an image download size and the ImageIO response contained an invalid content-length header (%s), so the download size cannot be determined",
)
}
i.size = contentLength
return nil
}
return newError(
EBug,
"the oVirt engine API did not return an image download size and the ImageIO response did not contain a content-length header, so the download size cannot be determined",
)
}
// Err will return the last error from the image download.
func (i *imageDownload) Err() error {
return i.lastError
}
// Initialized will return a channel that is closed when the image transfer is initialized.
func (i *imageDownload) Initialized() <-chan struct{} {
return i.done
}
// Read waits for the transfer to be properly initialized and in the transferring state, then
// reads from the HTTP response body. When there are no more bytes left it attempts to automatically
// finalize the transfer by calling the Close function.
func (i *imageDownload) Read(p []byte) (n int, err error) {
<-i.done
if i.lastError != nil {
return 0, i.lastError
}
n, err = i.reader.Read(p)
i.lock.Lock()
defer i.lock.Unlock()
i.bytesRead += uint64(n)
if i.bytesRead == i.size {
go func() {
_ = i.Close()
}()
}
return n, err
}
// Close is an implementation of the required function in io.ReadCloser and is responsible for
// closing the HTTP reader body and finalizing the image download. This is important so the disk
// does not stay locked.
func (i *imageDownload) Close() error {
i.lock.Lock()
defer i.lock.Unlock()
i.cancel()
if i.reader != nil {
err := i.reader.Close()
if err == nil {
i.reader = nil
}
return i.transfer.finalize(err)
}
return nil
}
// BytesRead returns the number of bytes already read from the download reader.
func (i *imageDownload) BytesRead() uint64 {
return i.bytesRead
}
// Size returns the size of the download. This does not necessarily match the disk size since the oVirt Engine may not
// return the disk size correctly.
func (i *imageDownload) Size() uint64 {
return i.size
}