#5294

Committed 11 May 2025 10:11AM UTC coverage: 59.288% (+0.05%) from 59.236%

Build # #5294

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

importer, util: Consolidate stream data to file (#3721)

* move util/file functions to importer/file

This contributes to the goal of eliminating util functions.
In future commits certain redundancy that existed between the importer
and util can be eliminated.

Signed-off-by: Adi Aloni <aaloni@redhat.com>

* importer replace streamDataToFile with StreamDataToFile

Previously there were two functions that were used to stream data to a
device:
- importer/util streamDataToFile
- importer/file StreamDataToFile

StreamDataToFile was originally introduced to handle preallocation for
host assisted clones[1]. Aside from that ability, the two functions are
identical in functionality.

This commit replaces streamDataToFile with StreamDataToFile.

[1] https://github.com/kubevirt/containerized-data-importer/pull/3352

Signed-off-by: Adi Aloni <aaloni@redhat.com>

* importer: respect preallocation in StreamDataToFile

Previously, the preallocation setting that's used in the data-processor
was only respected in the upload-server's clone-processor.

With this change, preallocation is respected in StreamDataToFile used
throughout all relevant datasources.

This should also result in more efficient imports as the default of the
preallocation setting is false.

Signed-off-by: Adi Aloni <aaloni@redhat.com>

* importer, errors: introduce IsNoCapacityError

Previously StreamDataToFile would assert out of space errors by matching
a substring in the error's message. This would not work in all
filesystems such as IBM's GPFS.

This commit introduces IsNoCapacityError which matches an error with all
insufficient capacity related errors (e.g., ENOSPC, EDQUOT) and uses it
instead of similar assertions.

Signed-off-by: Adi Aloni <aaloni@redhat.com>

---------

Signed-off-by: Adi Aloni <aaloni@redhat.com>

Run Details

79 of 129 new or added lines in 17 files covered. (61.24%)

2 existing lines in 1 file now uncovered.

16862 of 28441 relevant lines covered (59.29%)

0.66 hits per line

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

63.59

/pkg/uploadserver/uploadserver.go

/*
 * This file is part of the CDI project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * Copyright 2018 Red Hat, Inc.
 *
 */

package uploadserver

import (
        "archive/tar"
        "context"
        "crypto/tls"
        "crypto/x509"
        "fmt"
        "io"
        "mime/multipart"
        "net"
        "net/http"
        "os"
        "strings"
        "sync"
        "time"

        "github.com/golang/snappy"
        "github.com/pkg/errors"

        "k8s.io/klog/v2"

        cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
        "kubevirt.io/containerized-data-importer/pkg/common"
        "kubevirt.io/containerized-data-importer/pkg/importer"
        "kubevirt.io/containerized-data-importer/pkg/util"
        cryptowatch "kubevirt.io/containerized-data-importer/pkg/util/tls-crypto-watch"
)

const (
        healthzPath = "/healthz"
)

type Config struct {
        Insecure    bool
        BindAddress string
        BindPort    int

        Destination string

        ServerKeyFile, ServerCertFile string
        ClientCertFile, ClientName    string

        ImageSize          string
        FilesystemOverhead float64
        Preallocation      bool

        Deadline *time.Time

        CryptoConfig cryptowatch.CryptoConfig
}

// RunResult is the result of the upload server run
type RunResult struct {
        CloneTarget          bool
        PreallocationApplied bool
        DeadlinePassed       bool
}

// UploadServer is the interface to uploadServerApp
type UploadServer interface {
        Run() (*RunResult, error)
}

type uploadServerApp struct {
        config               *Config
        mux                  *http.ServeMux
        uploading            bool
        processing           bool
        done                 bool
        preallocationApplied bool
        cloneTarget          bool
        doneChan             chan struct{}
        errChan              chan error
        mutex                sync.Mutex
}

type imageReadCloser func(*http.Request) (io.ReadCloser, error)

// may be overridden in tests
var uploadProcessorFunc = newUploadStreamProcessor
var uploadProcessorFuncAsync = newAsyncUploadStreamProcessor

func bodyReadCloser(r *http.Request) (io.ReadCloser, error) {
        return r.Body, nil
}

func formReadCloser(r *http.Request) (io.ReadCloser, error) {
        multiReader, err := r.MultipartReader()
        if err != nil {
                return nil, err
        }

        var filePart *multipart.Part

        for {
                filePart, err = multiReader.NextPart()
                if err != nil || filePart.FormName() == "file" {
                        break
                }
                klog.Infof("Ignoring part %s", filePart.FormName())
        }

        // multiReader.NextPart() returns io.EOF when read everything
        if err != nil {
                return nil, err
        }

        return filePart, nil
}

func isCloneTarget(contentType string) bool {
        return contentType == common.BlockdeviceClone || contentType == common.FilesystemCloneContentType
}

// NewUploadServer returns a new instance of uploadServerApp
func NewUploadServer(config *Config) UploadServer {
        server := &uploadServerApp{
                config:    config,
                mux:       http.NewServeMux(),
                uploading: false,
                done:      false,
                doneChan:  make(chan struct{}),
                errChan:   make(chan error),
        }

        server.mux.HandleFunc(healthzPath, server.healthzHandler)
        for _, path := range common.SyncUploadPaths {
                server.mux.HandleFunc(path, server.uploadHandler(bodyReadCloser))
        }
        for _, path := range common.AsyncUploadPaths {
                server.mux.HandleFunc(path, server.uploadHandlerAsync(bodyReadCloser))
        }
        for _, path := range common.ArchiveUploadPaths {
                server.mux.HandleFunc(path, server.uploadArchiveHandler(bodyReadCloser))
        }
        for _, path := range common.SyncUploadFormPaths {
                server.mux.HandleFunc(path, server.uploadHandler(formReadCloser))
        }
        for _, path := range common.AsyncUploadFormPaths {
                server.mux.HandleFunc(path, server.uploadHandlerAsync(formReadCloser))
        }

        return server
}

func (app *uploadServerApp) Run() (*RunResult, error) {
        uploadServer := http.Server{
                Handler:           app,
                ReadHeaderTimeout: 10 * time.Second,
        }

        uploadListener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", app.config.BindAddress, app.config.BindPort))
        if err != nil {
                return nil, errors.Wrap(err, "Error creating upload listerner")
        }

        tlsConfig, err := app.getTLSConfig()
        if err != nil {
                return nil, errors.Wrap(err, "Error getting TLS config")
        }

        go func() {
                defer uploadListener.Close()

                // maybe bind port was 0 (unit tests) assign port here
                app.config.BindPort = uploadListener.Addr().(*net.TCPAddr).Port

                if tlsConfig != nil {
                        uploadServer.TLSConfig = tlsConfig
                        app.errChan <- uploadServer.ServeTLS(uploadListener, "", "")
                        return
                }

                // not sure we want to support this code path
                app.errChan <- uploadServer.Serve(uploadListener)
        }()

        var timeChan <-chan time.Time

        if app.config.Deadline != nil {
                timeChan = time.After(time.Until(*app.config.Deadline))
        } else {
                tc := make(chan time.Time)
                defer close(tc)
                timeChan = tc
        }

        select {
        case err = <-app.errChan:
                if err != nil {
                        klog.Errorf("HTTP server returned error %s", err.Error())
                        return nil, err
                }
        case <-app.doneChan:
                klog.Info("Shutting down http server after successful upload")
                if err := uploadServer.Shutdown(context.Background()); err != nil {
                        klog.Errorf("failed to shutdown uploadServer; %v", err)
                }
        case <-timeChan:
                klog.Info("deadline exceeded, shutting down")
                app.mutex.Lock()
                defer app.mutex.Unlock()
                for {
                        if app.uploading || app.processing {
                                klog.Info("waiting for upload to finish")
                                app.mutex.Unlock()
                                time.Sleep(2 * time.Second)
                                app.mutex.Lock()
                        } else {
                                break
                        }
                }
                if !app.done {
                        klog.Info("upload not done, process exiting")
                        return &RunResult{DeadlinePassed: true}, nil
                }
        }

        result := &RunResult{
                CloneTarget:          app.cloneTarget,
                PreallocationApplied: app.preallocationApplied,
        }

        return result, nil
}

func (app *uploadServerApp) getTLSConfig() (*tls.Config, error) {
        if app.config.ServerCertFile == "" || app.config.ServerKeyFile == "" {
                if !app.config.Insecure {
                        return nil, errors.New("invalid TLS config")
                }
                return nil, nil
        }

        //nolint:gosec // False positive: Min version is not known statically
        config := &tls.Config{
                CipherSuites: app.config.CryptoConfig.CipherSuites,
                ClientAuth:   tls.VerifyClientCertIfGiven,
                MinVersion:   app.config.CryptoConfig.MinVersion,
        }

        if app.config.ClientCertFile != "" {
                bs, err := os.ReadFile(app.config.ClientCertFile)
                if err != nil {
                        return nil, err
                }

                caCertPool := x509.NewCertPool()
                if ok := caCertPool.AppendCertsFromPEM(bs); !ok {
                        return nil, err
                }

                config.ClientCAs = caCertPool
        }

        cert, err := tls.LoadX509KeyPair(app.config.ServerCertFile, app.config.ServerKeyFile)
        if err != nil {
                return nil, err
        }

        config.Certificates = []tls.Certificate{cert}

        return config, nil
}

func (app *uploadServerApp) ServeHTTP(w http.ResponseWriter, r *http.Request) {
        app.mux.ServeHTTP(w, r)
}

func (app *uploadServerApp) healthzHandler(w http.ResponseWriter, r *http.Request) {
        if _, err := io.WriteString(w, "OK"); err != nil {
                klog.Errorf("healthzHandler: failed to send response; %v", err)
        }
}

func (app *uploadServerApp) validateShouldHandleRequest(w http.ResponseWriter, r *http.Request) bool {
        if r.Method != http.MethodPost {
                w.WriteHeader(http.StatusNotFound)
                return false
        }

        if r.TLS != nil {
                if len(r.TLS.VerifiedChains) == 0 {
                        w.WriteHeader(http.StatusUnauthorized)
                        return false
                }

                found := false
                for _, cert := range r.TLS.PeerCertificates {
                        if cert.Subject.CommonName == app.config.ClientName {
                                found = true
                                break
                        }
                }

                if !found {
                        w.WriteHeader(http.StatusUnauthorized)
                        return false
                }
        } else {
                if !app.config.Insecure {
                        w.WriteHeader(http.StatusUnauthorized)
                        return false
                }
                klog.V(3).Infof("Handling HTTP connection")
        }

        app.mutex.Lock()
        defer app.mutex.Unlock()

        if app.uploading || app.processing {
                klog.Warning("Got concurrent upload request")
                w.WriteHeader(http.StatusServiceUnavailable)
                return false
        }

        if app.done {
                klog.Warning("Got upload request after already done")
                w.WriteHeader(http.StatusConflict)
                return false
        }

        app.uploading = true

        return true
}

func (app *uploadServerApp) uploadHandlerAsync(irc imageReadCloser) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
                if r.Method == http.MethodHead {
                        w.WriteHeader(http.StatusOK)
                        return
                }

                if !app.validateShouldHandleRequest(w, r) {
                        return
                }

                cdiContentType := r.Header.Get(common.UploadContentTypeHeader)

                klog.Infof("Content type header is %q\n", cdiContentType)

                readCloser, err := irc(r)
                if err != nil {
                        w.WriteHeader(http.StatusBadRequest)
                }

                processor, err := uploadProcessorFuncAsync(readCloser, app.config.Destination, app.config.ImageSize, app.config.FilesystemOverhead, app.config.Preallocation, cdiContentType)

                app.mutex.Lock()
                defer app.mutex.Unlock()
                app.uploading = false

                if err != nil {
                        handleStreamError(w, err)
                        return
                }

                app.processing = true

                // Start processing.
                go func() {
                        err := processor.ProcessDataResume()
                        app.mutex.Lock()
                        defer app.mutex.Unlock()
                        app.processing = false
                        if err != nil {
                                klog.Errorf("Error during resumed processing: %v", err)
                                app.errChan <- err
                                return
                        }
                        defer close(app.doneChan)
                        app.done = true
                        app.preallocationApplied = processor.PreallocationApplied()
                        app.cloneTarget = isCloneTarget(cdiContentType)
                        klog.Infof("Wrote data to %s", app.config.Destination)
                }()

                klog.Info("Returning success to caller, continue processing in background")
        }
}

func (app *uploadServerApp) processUpload(irc imageReadCloser, w http.ResponseWriter, r *http.Request, dvContentType cdiv1.DataVolumeContentType) {
        if !app.validateShouldHandleRequest(w, r) {
                return
        }

        cdiContentType := r.Header.Get(common.UploadContentTypeHeader)

        klog.Infof("Content type header is %q\n", cdiContentType)

        readCloser, err := irc(r)
        if err != nil {
                w.WriteHeader(http.StatusBadRequest)
        }

        preallocationApplied, err := uploadProcessorFunc(readCloser, app.config.Destination, app.config.ImageSize, app.config.FilesystemOverhead, app.config.Preallocation, cdiContentType, dvContentType)

        app.mutex.Lock()
        defer app.mutex.Unlock()
        app.uploading = false

        if err != nil {
                handleStreamError(w, err)
                return
        }

        app.done = true
        app.preallocationApplied = preallocationApplied
        app.cloneTarget = isCloneTarget(cdiContentType)
        close(app.doneChan)

        if dvContentType == cdiv1.DataVolumeArchive {
                klog.Infof("Wrote archive data")
        } else {
                klog.Infof("Wrote data to %s", app.config.Destination)
        }
}

func (app *uploadServerApp) uploadHandler(irc imageReadCloser) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
                app.processUpload(irc, w, r, cdiv1.DataVolumeKubeVirt)
        }
}

func (app *uploadServerApp) uploadArchiveHandler(irc imageReadCloser) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
                app.processUpload(irc, w, r, cdiv1.DataVolumeArchive)
        }
}

func newAsyncUploadStreamProcessor(stream io.ReadCloser, dest, imageSize string, filesystemOverhead float64, preallocation bool, sourceContentType string) (*importer.DataProcessor, error) {
        if isCloneTarget(sourceContentType) {
                return nil, fmt.Errorf("async clone not supported")
        }

        uds := importer.NewAsyncUploadDataSource(newContentReader(stream, sourceContentType))
        processor := importer.NewDataProcessor(uds, dest, common.ImporterVolumePath, common.ScratchDataDir, imageSize, filesystemOverhead, preallocation, "")
        return processor, processor.ProcessDataWithPause()
}

func newUploadStreamProcessor(stream io.ReadCloser, dest, imageSize string, filesystemOverhead float64, preallocation bool, sourceContentType string, dvContentType cdiv1.DataVolumeContentType) (bool, error) {
        stream = newContentReader(stream, sourceContentType)
        if isCloneTarget(sourceContentType) {
                return cloneProcessor(stream, sourceContentType, dest, preallocation)
        }

        // Clone block device to block device or file system
        uds := importer.NewUploadDataSource(stream, dvContentType)
        processor := importer.NewDataProcessor(uds, dest, common.ImporterVolumePath, common.ScratchDataDir, imageSize, filesystemOverhead, preallocation, "")
        err := processor.ProcessData()
        return processor.PreallocationApplied(), err
}

func cloneProcessor(stream io.ReadCloser, contentType, dest string, preallocate bool) (bool, error) {
        if contentType == common.FilesystemCloneContentType {
                if dest != common.WriteBlockPath {
                        return fileToFileCloneProcessor(stream)
                }

                tarImageReader, err := newTarDiskImageReader(stream)
                if err != nil {
                        stream.Close()
                        return false, err
                }
                stream = tarImageReader
        }

        defer stream.Close()

        _, _, err := importer.StreamDataToFile(stream, dest, preallocate)
        if err != nil {
                return false, err
        }

        return false, nil
}

func fileToFileCloneProcessor(stream io.ReadCloser) (bool, error) {
        defer stream.Close()
        if err := util.UnArchiveTar(stream, common.ImporterVolumePath); err != nil {
                return false, errors.Wrapf(err, "error unarchiving to %s", common.ImporterVolumePath)
        }
        return true, nil
}

type closeWrapper struct {
        io.Reader
        closers []io.Closer
}

func (c *closeWrapper) Close() error {
        var err error
        for _, closer := range c.closers {
                if e := closer.Close(); e != nil {
                        err = e
                }
        }
        return err
}

type tarDiskImageReader struct {
        tr           *tar.Reader
        size, offset int64
}

func (r *tarDiskImageReader) Read(p []byte) (int, error) {
        if r.offset >= r.size {
                return 0, io.EOF
        }
        remaining := r.size - r.offset
        if int(remaining) < len(p) {
                p = p[:remaining]
        }
        n, err := r.tr.Read(p)
        r.offset += int64(n)
        klog.V(3).Infof("Read %d bytes, offset %d, size %d", n, r.offset, r.size)
        return n, err
}

func newTarDiskImageReader(stream io.ReadCloser) (io.ReadCloser, error) {
        tr := tar.NewReader(stream)
        for {
                header, err := tr.Next()
                if err == io.EOF {
                        break
                }
                if err != nil {
                        return nil, err
                }
                if !strings.Contains(header.Name, common.DiskImageName) {
                        continue
                }
                return &closeWrapper{
                        Reader:  &tarDiskImageReader{tr: tr, size: header.Size},
                        closers: []io.Closer{stream},
                }, nil
        }
        return nil, fmt.Errorf("no disk image found in tar")
}

func newContentReader(stream io.ReadCloser, contentType string) io.ReadCloser {
        if isCloneTarget(contentType) {
                return newSnappyReadCloser(stream)
        }
        return stream
}

func newSnappyReadCloser(stream io.ReadCloser) io.ReadCloser {
        return &closeWrapper{
                Reader:  snappy.NewReader(stream),
                closers: []io.Closer{stream},
        }
}

func handleStreamError(w http.ResponseWriter, err error) {
        if importer.IsNoCapacityError(err) {
                w.WriteHeader(http.StatusBadRequest)
                err = errors.New("effective image size is larger than the reported available storage. A larger PVC is required")
        } else {
                w.WriteHeader(http.StatusInternalServerError)
        }
        klog.Errorf("Saving stream failed: %s", err)

        _, writeErr := fmt.Fprintf(w, "Saving stream failed: %s", err.Error())
        if writeErr != nil {
                klog.Errorf("failed to send response; %v", err)
        }
}

1	/*
2	* This file is part of the CDI project
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*
16	* Copyright 2018 Red Hat, Inc.
17	*
18	*/
19
20	package uploadserver
21
22	import (
23	"archive/tar"
24	"context"
25	"crypto/tls"
26	"crypto/x509"
27	"fmt"
28	"io"
29	"mime/multipart"
30	"net"
31	"net/http"
32	"os"
33	"strings"
34	"sync"
35	"time"
36
37	"github.com/golang/snappy"
38	"github.com/pkg/errors"
39
40	"k8s.io/klog/v2"
41
42	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
43	"kubevirt.io/containerized-data-importer/pkg/common"
44	"kubevirt.io/containerized-data-importer/pkg/importer"
45	"kubevirt.io/containerized-data-importer/pkg/util"
46	cryptowatch "kubevirt.io/containerized-data-importer/pkg/util/tls-crypto-watch"
47	)
48
49	const (
50	healthzPath = "/healthz"
51	)
52
53	type Config struct {
54	Insecure bool
55	BindAddress string
56	BindPort int
57
58	Destination string
59
60	ServerKeyFile, ServerCertFile string
61	ClientCertFile, ClientName string
62
63	ImageSize string
64	FilesystemOverhead float64
65	Preallocation bool
66
67	Deadline *time.Time
68
69	CryptoConfig cryptowatch.CryptoConfig
70	}
71
72	// RunResult is the result of the upload server run
73	type RunResult struct {
74	CloneTarget bool
75	PreallocationApplied bool
76	DeadlinePassed bool
77	}
78
79	// UploadServer is the interface to uploadServerApp
80	type UploadServer interface {
81	Run() (*RunResult, error)
82	}
83
84	type uploadServerApp struct {
85	config *Config
86	mux *http.ServeMux
87	uploading bool
88	processing bool
89	done bool
90	preallocationApplied bool
91	cloneTarget bool
92	doneChan chan struct{}
93	errChan chan error
94	mutex sync.Mutex
95	}
96
97	type imageReadCloser func(*http.Request) (io.ReadCloser, error)
98
99	// may be overridden in tests
100	var uploadProcessorFunc = newUploadStreamProcessor
101	var uploadProcessorFuncAsync = newAsyncUploadStreamProcessor
102
103	func bodyReadCloser(r *http.Request) (io.ReadCloser, error) {	1✔
104	return r.Body, nil	1✔
105	}	1✔
106
107	func formReadCloser(r *http.Request) (io.ReadCloser, error) {	1✔
108	multiReader, err := r.MultipartReader()	1✔
109	if err != nil {	1✔
110	return nil, err	×
111	}	×
112
113	var filePart *multipart.Part	1✔
114		1✔
115	for {	2✔
116	filePart, err = multiReader.NextPart()	1✔
117	if err != nil \|\| filePart.FormName() == "file" {	2✔
118	break	1✔
119	}
120	klog.Infof("Ignoring part %s", filePart.FormName())	×
121	}
122
123	// multiReader.NextPart() returns io.EOF when read everything
124	if err != nil {	1✔
125	return nil, err	×
126	}	×
127
128	return filePart, nil	1✔
129	}
130
131	func isCloneTarget(contentType string) bool {	1✔
132	return contentType == common.BlockdeviceClone \|\| contentType == common.FilesystemCloneContentType	1✔
133	}	1✔
134
135	// NewUploadServer returns a new instance of uploadServerApp
136	func NewUploadServer(config *Config) UploadServer {	1✔
137	server := &uploadServerApp{	1✔
138	config: config,	1✔
139	mux: http.NewServeMux(),	1✔
140	uploading: false,	1✔
141	done: false,	1✔
142	doneChan: make(chan struct{}),	1✔
143	errChan: make(chan error),	1✔
144	}	1✔
145		1✔
146	server.mux.HandleFunc(healthzPath, server.healthzHandler)	1✔
147	for _, path := range common.SyncUploadPaths {	2✔
148	server.mux.HandleFunc(path, server.uploadHandler(bodyReadCloser))	1✔
149	}	1✔
150	for _, path := range common.AsyncUploadPaths {	2✔
151	server.mux.HandleFunc(path, server.uploadHandlerAsync(bodyReadCloser))	1✔
152	}	1✔
153	for _, path := range common.ArchiveUploadPaths {	2✔
154	server.mux.HandleFunc(path, server.uploadArchiveHandler(bodyReadCloser))	1✔
155	}	1✔
156	for _, path := range common.SyncUploadFormPaths {	2✔
157	server.mux.HandleFunc(path, server.uploadHandler(formReadCloser))	1✔
158	}	1✔
159	for _, path := range common.AsyncUploadFormPaths {	2✔
160	server.mux.HandleFunc(path, server.uploadHandlerAsync(formReadCloser))	1✔
161	}	1✔
162
163	return server	1✔
164	}
165
166	func (app uploadServerApp) Run() (RunResult, error) {	1✔
167	uploadServer := http.Server{	1✔
168	Handler: app,	1✔
169	ReadHeaderTimeout: 10 * time.Second,	1✔
170	}	1✔
171		1✔
172	uploadListener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", app.config.BindAddress, app.config.BindPort))	1✔
173	if err != nil {	1✔
174	return nil, errors.Wrap(err, "Error creating upload listerner")	×
175	}	×
176
177	tlsConfig, err := app.getTLSConfig()	1✔
178	if err != nil {	1✔
179	return nil, errors.Wrap(err, "Error getting TLS config")	×
180	}	×
181
182	go func() {	2✔
183	defer uploadListener.Close()	1✔
184		1✔
185	// maybe bind port was 0 (unit tests) assign port here	1✔
186	app.config.BindPort = uploadListener.Addr().(*net.TCPAddr).Port	1✔
187		1✔
188	if tlsConfig != nil {	2✔
189	uploadServer.TLSConfig = tlsConfig	1✔
190	app.errChan <- uploadServer.ServeTLS(uploadListener, "", "")	1✔
191	return	1✔
192	}	1✔
193
194	// not sure we want to support this code path
195	app.errChan <- uploadServer.Serve(uploadListener)	×
196	}()
197
198	var timeChan <-chan time.Time	1✔
199		1✔
200	if app.config.Deadline != nil {	2✔
201	timeChan = time.After(time.Until(*app.config.Deadline))	1✔
202	} else {	2✔
203	tc := make(chan time.Time)	1✔
204	defer close(tc)	1✔
205	timeChan = tc	1✔
206	}	1✔
207
208	select {	1✔
209	case err = <-app.errChan:	×
210	if err != nil {	×
211	klog.Errorf("HTTP server returned error %s", err.Error())	×
212	return nil, err	×
213	}	×
214	case <-app.doneChan:	1✔
215	klog.Info("Shutting down http server after successful upload")	1✔
216	if err := uploadServer.Shutdown(context.Background()); err != nil {	1✔
217	klog.Errorf("failed to shutdown uploadServer; %v", err)	×
218	}	×
219	case <-timeChan:	1✔
220	klog.Info("deadline exceeded, shutting down")	1✔
221	app.mutex.Lock()	1✔
222	defer app.mutex.Unlock()	1✔
223	for {	2✔
224	if app.uploading \|\| app.processing {	1✔
225	klog.Info("waiting for upload to finish")	×
226	app.mutex.Unlock()	×
227	time.Sleep(2 * time.Second)	×
228	app.mutex.Lock()	×
229	} else {	1✔
230	break	1✔
231	}
232	}
233	if !app.done {	2✔
234	klog.Info("upload not done, process exiting")	1✔
235	return &RunResult{DeadlinePassed: true}, nil	1✔
236	}	1✔
237	}
238
239	result := &RunResult{	1✔
240	CloneTarget: app.cloneTarget,	1✔
241	PreallocationApplied: app.preallocationApplied,	1✔
242	}	1✔
243		1✔
244	return result, nil	1✔
245	}
246
247	func (app uploadServerApp) getTLSConfig() (tls.Config, error) {	1✔
248	if app.config.ServerCertFile == "" \|\| app.config.ServerKeyFile == "" {	1✔
249	if !app.config.Insecure {	×
250	return nil, errors.New("invalid TLS config")	×
251	}	×
252	return nil, nil	×
253	}
254
255	//nolint:gosec // False positive: Min version is not known statically
256	config := &tls.Config{	1✔
257	CipherSuites: app.config.CryptoConfig.CipherSuites,	1✔
258	ClientAuth: tls.VerifyClientCertIfGiven,	1✔
259	MinVersion: app.config.CryptoConfig.MinVersion,	1✔
260	}	1✔
261		1✔
262	if app.config.ClientCertFile != "" {	2✔
263	bs, err := os.ReadFile(app.config.ClientCertFile)	1✔
264	if err != nil {	1✔
265	return nil, err	×
266	}	×
267
268	caCertPool := x509.NewCertPool()	1✔
269	if ok := caCertPool.AppendCertsFromPEM(bs); !ok {	1✔
270	return nil, err	×
271	}	×
272
273	config.ClientCAs = caCertPool	1✔
274	}
275
276	cert, err := tls.LoadX509KeyPair(app.config.ServerCertFile, app.config.ServerKeyFile)	1✔
277	if err != nil {	1✔
278	return nil, err	×
279	}	×
280
281	config.Certificates = []tls.Certificate{cert}	1✔
282		1✔
283	return config, nil	1✔
284	}
285
286	func (app uploadServerApp) ServeHTTP(w http.ResponseWriter, r http.Request) {	1✔
287	app.mux.ServeHTTP(w, r)	1✔
288	}	1✔
289
290	func (app uploadServerApp) healthzHandler(w http.ResponseWriter, r http.Request) {	1✔
291	if _, err := io.WriteString(w, "OK"); err != nil {	1✔
292	klog.Errorf("healthzHandler: failed to send response; %v", err)	×
293	}	×
294	}
295
296	func (app uploadServerApp) validateShouldHandleRequest(w http.ResponseWriter, r http.Request) bool {	1✔
297	if r.Method != http.MethodPost {	2✔
298	w.WriteHeader(http.StatusNotFound)	1✔
299	return false	1✔
300	}	1✔
301
302	if r.TLS != nil {	2✔
303	if len(r.TLS.VerifiedChains) == 0 {	1✔
304	w.WriteHeader(http.StatusUnauthorized)	×
305	return false	×
306	}	×
307
308	found := false	1✔
309	for _, cert := range r.TLS.PeerCertificates {	2✔
310	if cert.Subject.CommonName == app.config.ClientName {	2✔
311	found = true	1✔
312	break	1✔
313	}
314	}
315
316	if !found {	2✔
317	w.WriteHeader(http.StatusUnauthorized)	1✔
318	return false	1✔
319	}	1✔
320	} else {	1✔
321	if !app.config.Insecure {	1✔
322	w.WriteHeader(http.StatusUnauthorized)	×
323	return false	×
324	}	×
325	klog.V(3).Infof("Handling HTTP connection")	1✔
326	}
327
328	app.mutex.Lock()	1✔
329	defer app.mutex.Unlock()	1✔
330		1✔
331	if app.uploading \|\| app.processing {	2✔
332	klog.Warning("Got concurrent upload request")	1✔
333	w.WriteHeader(http.StatusServiceUnavailable)	1✔
334	return false	1✔
335	}	1✔
336
337	if app.done {	2✔
338	klog.Warning("Got upload request after already done")	1✔
339	w.WriteHeader(http.StatusConflict)	1✔
340	return false	1✔
341	}	1✔
342
343	app.uploading = true	1✔
344		1✔
345	return true	1✔
346	}
347
348	func (app *uploadServerApp) uploadHandlerAsync(irc imageReadCloser) http.HandlerFunc {	1✔
349	return func(w http.ResponseWriter, r *http.Request) {	2✔
350	if r.Method == http.MethodHead {	2✔
351	w.WriteHeader(http.StatusOK)	1✔
352	return	1✔
353	}	1✔
354
355	if !app.validateShouldHandleRequest(w, r) {	2✔
356	return	1✔
357	}	1✔
358
359	cdiContentType := r.Header.Get(common.UploadContentTypeHeader)	1✔
360		1✔
361	klog.Infof("Content type header is %q\n", cdiContentType)	1✔
362		1✔
363	readCloser, err := irc(r)	1✔
364	if err != nil {	1✔
365	w.WriteHeader(http.StatusBadRequest)	×
366	}	×
367
368	processor, err := uploadProcessorFuncAsync(readCloser, app.config.Destination, app.config.ImageSize, app.config.FilesystemOverhead, app.config.Preallocation, cdiContentType)	1✔
369		1✔
370	app.mutex.Lock()	1✔
371	defer app.mutex.Unlock()	1✔
372	app.uploading = false	1✔
373		1✔
374	if err != nil {	2✔
375	handleStreamError(w, err)	1✔
376	return	1✔
377	}	1✔
378
379	app.processing = true	1✔
380		1✔
381	// Start processing.	1✔
382	go func() {	2✔
383	err := processor.ProcessDataResume()	1✔
384	app.mutex.Lock()	1✔
385	defer app.mutex.Unlock()	1✔
386	app.processing = false	1✔
387	if err != nil {	1✔
388	klog.Errorf("Error during resumed processing: %v", err)	×
389	app.errChan <- err	×
390	return	×
391	}	×
392	defer close(app.doneChan)	1✔
393	app.done = true	1✔
394	app.preallocationApplied = processor.PreallocationApplied()	1✔
395	app.cloneTarget = isCloneTarget(cdiContentType)	1✔
396	klog.Infof("Wrote data to %s", app.config.Destination)	1✔
397	}()
398
399	klog.Info("Returning success to caller, continue processing in background")	1✔
400	}
401	}
402
403	func (app uploadServerApp) processUpload(irc imageReadCloser, w http.ResponseWriter, r http.Request, dvContentType cdiv1.DataVolumeContentType) {	1✔
404	if !app.validateShouldHandleRequest(w, r) {	2✔
405	return	1✔
406	}	1✔
407
408	cdiContentType := r.Header.Get(common.UploadContentTypeHeader)	1✔
409		1✔
410	klog.Infof("Content type header is %q\n", cdiContentType)	1✔
411		1✔
412	readCloser, err := irc(r)	1✔
413	if err != nil {	1✔
414	w.WriteHeader(http.StatusBadRequest)	×
415	}	×
416
417	preallocationApplied, err := uploadProcessorFunc(readCloser, app.config.Destination, app.config.ImageSize, app.config.FilesystemOverhead, app.config.Preallocation, cdiContentType, dvContentType)	1✔
418		1✔
419	app.mutex.Lock()	1✔
420	defer app.mutex.Unlock()	1✔
421	app.uploading = false	1✔
422		1✔
423	if err != nil {	2✔
424	handleStreamError(w, err)	1✔
425	return	1✔
426	}	1✔
427
428	app.done = true	1✔
429	app.preallocationApplied = preallocationApplied	1✔
430	app.cloneTarget = isCloneTarget(cdiContentType)	1✔
431	close(app.doneChan)	1✔
432		1✔
433	if dvContentType == cdiv1.DataVolumeArchive {	1✔
434	klog.Infof("Wrote archive data")	×
435	} else {	1✔
436	klog.Infof("Wrote data to %s", app.config.Destination)	1✔
437	}	1✔
438	}
439
440	func (app *uploadServerApp) uploadHandler(irc imageReadCloser) http.HandlerFunc {	1✔
441	return func(w http.ResponseWriter, r *http.Request) {	2✔
442	app.processUpload(irc, w, r, cdiv1.DataVolumeKubeVirt)	1✔
443	}	1✔
444	}
445
446	func (app *uploadServerApp) uploadArchiveHandler(irc imageReadCloser) http.HandlerFunc {	1✔
447	return func(w http.ResponseWriter, r *http.Request) {	2✔
448	app.processUpload(irc, w, r, cdiv1.DataVolumeArchive)	1✔
449	}	1✔
450	}
451
452	func newAsyncUploadStreamProcessor(stream io.ReadCloser, dest, imageSize string, filesystemOverhead float64, preallocation bool, sourceContentType string) (*importer.DataProcessor, error) {	×
453	if isCloneTarget(sourceContentType) {	×
454	return nil, fmt.Errorf("async clone not supported")	×
455	}	×
456
457	uds := importer.NewAsyncUploadDataSource(newContentReader(stream, sourceContentType))	×
458	processor := importer.NewDataProcessor(uds, dest, common.ImporterVolumePath, common.ScratchDataDir, imageSize, filesystemOverhead, preallocation, "")	×
459	return processor, processor.ProcessDataWithPause()	×
460	}
461
462	func newUploadStreamProcessor(stream io.ReadCloser, dest, imageSize string, filesystemOverhead float64, preallocation bool, sourceContentType string, dvContentType cdiv1.DataVolumeContentType) (bool, error) {	×
463	stream = newContentReader(stream, sourceContentType)	×
464	if isCloneTarget(sourceContentType) {	×
465	return cloneProcessor(stream, sourceContentType, dest, preallocation)	×
466	}	×
467
468	// Clone block device to block device or file system
469	uds := importer.NewUploadDataSource(stream, dvContentType)	×
470	processor := importer.NewDataProcessor(uds, dest, common.ImporterVolumePath, common.ScratchDataDir, imageSize, filesystemOverhead, preallocation, "")	×
471	err := processor.ProcessData()	×
472	return processor.PreallocationApplied(), err	×
473	}
474
475	func cloneProcessor(stream io.ReadCloser, contentType, dest string, preallocate bool) (bool, error) {	×
476	if contentType == common.FilesystemCloneContentType {	×
477	if dest != common.WriteBlockPath {	×
478	return fileToFileCloneProcessor(stream)	×
479	}	×
480
481	tarImageReader, err := newTarDiskImageReader(stream)	×
482	if err != nil {	×
483	stream.Close()	×
484	return false, err	×
485	}	×
486	stream = tarImageReader	×
487	}
488
489	defer stream.Close()	×
NEW 490		×
NEW 491	_, _, err := importer.StreamDataToFile(stream, dest, preallocate)	×
492	if err != nil {	×
493	return false, err	×
494	}	×
495
496	return false, nil	×
497	}
498
499	func fileToFileCloneProcessor(stream io.ReadCloser) (bool, error) {	×
500	defer stream.Close()	×
501	if err := util.UnArchiveTar(stream, common.ImporterVolumePath); err != nil {	×
502	return false, errors.Wrapf(err, "error unarchiving to %s", common.ImporterVolumePath)	×
503	}	×
504	return true, nil	×
505	}
506
507	type closeWrapper struct {
508	io.Reader
509	closers []io.Closer
510	}
511
512	func (c *closeWrapper) Close() error {	×
513	var err error	×
514	for _, closer := range c.closers {	×
515	if e := closer.Close(); e != nil {	×
516	err = e	×
517	}	×
518	}
519	return err	×
520	}
521
522	type tarDiskImageReader struct {
523	tr *tar.Reader
524	size, offset int64
525	}
526
527	func (r *tarDiskImageReader) Read(p []byte) (int, error) {	×
528	if r.offset >= r.size {	×
529	return 0, io.EOF	×
530	}	×
531	remaining := r.size - r.offset	×
532	if int(remaining) < len(p) {	×
533	p = p[:remaining]	×
534	}	×
535	n, err := r.tr.Read(p)	×
536	r.offset += int64(n)	×
537	klog.V(3).Infof("Read %d bytes, offset %d, size %d", n, r.offset, r.size)	×
538	return n, err	×
539	}
540
541	func newTarDiskImageReader(stream io.ReadCloser) (io.ReadCloser, error) {	×
542	tr := tar.NewReader(stream)	×
543	for {	×
544	header, err := tr.Next()	×
545	if err == io.EOF {	×
546	break	×
547	}
548	if err != nil {	×
549	return nil, err	×
550	}	×
551	if !strings.Contains(header.Name, common.DiskImageName) {	×
552	continue	×
553	}
554	return &closeWrapper{	×
555	Reader: &tarDiskImageReader{tr: tr, size: header.Size},	×
556	closers: []io.Closer{stream},	×
557	}, nil	×
558	}
559	return nil, fmt.Errorf("no disk image found in tar")	×
560	}
561
562	func newContentReader(stream io.ReadCloser, contentType string) io.ReadCloser {	×
563	if isCloneTarget(contentType) {	×
564	return newSnappyReadCloser(stream)	×
565	}	×
566	return stream	×
567	}
568
569	func newSnappyReadCloser(stream io.ReadCloser) io.ReadCloser {	×
570	return &closeWrapper{	×
571	Reader: snappy.NewReader(stream),	×
572	closers: []io.Closer{stream},	×
573	}	×
574	}	×
575
576	func handleStreamError(w http.ResponseWriter, err error) {	1✔
577	if importer.IsNoCapacityError(err) {	1✔
578	w.WriteHeader(http.StatusBadRequest)	×
579	err = errors.New("effective image size is larger than the reported available storage. A larger PVC is required")	×
580	} else {	1✔
581	w.WriteHeader(http.StatusInternalServerError)	1✔
582	}	1✔
583	klog.Errorf("Saving stream failed: %s", err)	1✔
584		1✔
585	_, writeErr := fmt.Fprintf(w, "Saving stream failed: %s", err.Error())	1✔
586	if writeErr != nil {	1✔
587	klog.Errorf("failed to send response; %v", err)	×
588	}	×
589	}

kubevirt / containerized-data-importer / #5294

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