951615729

pending completion

Build # 951615729

Build Type

push

gitlab-ci

Committed by

larsewi

Commit Message

fix: Add 'build/' to '.gitignore'

Ticket: None
Changelog: None
Signed-off-by: Lars Erik Wik <lars.erik.wik@northern.tech>

Run Details

4199 of 5286 relevant lines covered (79.44%)

166.43 hits per line

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

84.04

/mender-update/update_module/v3/update_module_download.cpp

// Copyright 2023 Northern.tech AS
//
//    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.

#include <mender-update/update_module/v3/update_module.hpp>

#include <common/events.hpp>
#include <common/events_io.hpp>
#include <common/log.hpp>
#include <common/path.hpp>

namespace mender {
namespace update {
namespace update_module {
namespace v3 {

namespace log = mender::common::log;
namespace path = mender::common::path;

void UpdateModule::StartDownloadProcess() {
        log::Debug(
                "Calling Update Module with command `" + update_module_path_ + " Download "
                + update_module_workdir_ + "`.");
        download_->proc_ = make_shared<processes::Process>(
                vector<string> {update_module_path_, "Download", update_module_workdir_});

        download_->proc_->SetWorkDir(update_module_workdir_);

        auto err = PrepareStreamNextPipe();
        if (err != error::NoError) {
                DownloadErrorHandler(err);
                return;
        }

        err = download_->proc_->Start();
        if (err != error::NoError) {
                DownloadErrorHandler(GetProcessError(err));
                return;
        }

        err = download_->proc_->AsyncWait(
                download_->event_loop_, [this](error::Error err) { ProcessEndedHandler(err); });
        if (err != error::NoError) {
                DownloadErrorHandler(err);
                return;
        }

        download_->proc_timeout_.AsyncWait(
                chrono::seconds(ctx_.GetConfig().module_timeout_seconds), [this](error::Error err) {
                        if (err != error::NoError) {
                                DownloadErrorHandler(
                                        err.WithContext("Error while waiting for Update Module Download process"));
                        } else {
                                DownloadTimeoutHandler();
                        }
                });

        DownloadErrorHandler(OpenStreamNextPipe(
                [this](io::ExpectedAsyncWriterPtr writer) { StreamNextOpenHandler(writer); }));
}

void UpdateModule::StreamNextOpenHandler(io::ExpectedAsyncWriterPtr writer) {
        if (!writer) {
                DownloadErrorHandler(writer.error());
                return;
        }
        download_->stream_next_writer_ = writer.value();

        download_->module_has_started_download_ = true;

        auto reader = download_->payload_.Next();
        if (!reader) {
                if (reader.error().code
                        == artifact::parser_error::MakeError(
                                   artifact::parser_error::NoMorePayloadFilesError, "")
                                   .code) {
                        download_->module_has_finished_download_ = true;
                        log::Debug("Update Module finished all downloads");
                        EndStreamNext();
                } else {
                        DownloadErrorHandler(reader.error());
                }
                return;
        }
        auto payload_reader = make_shared<artifact::Reader>(std::move(reader.value()));
        download_->current_payload_reader_ =
                make_shared<events::io::AsyncReaderFromReader>(download_->event_loop_, payload_reader);
        download_->current_payload_name_ = payload_reader->Name();

        auto stream_path =
                path::Join(update_module_workdir_, string("streams"), download_->current_payload_name_);
        DownloadErrorHandler(PrepareAndOpenStreamPipe(
                stream_path, [this](io::ExpectedAsyncWriterPtr writer) { StreamOpenHandler(writer); }));

        string stream_next_string = path::Join("streams", download_->current_payload_name_);
        size_t entry_size = stream_next_string.size() + 1;
        if (entry_size > download_->buffer_.size()) {
                DownloadErrorHandler(error::Error(
                        make_error_condition(errc::no_buffer_space), "Payload name is too large for buffer"));
                return;
        }
        copy(stream_next_string.begin(), stream_next_string.end(), download_->buffer_.begin());
        download_->buffer_[entry_size - 1] = '\n';
        DownloadErrorHandler(download_->stream_next_writer_->AsyncWrite(
                download_->buffer_.begin(),
                download_->buffer_.begin() + entry_size,
                [this, entry_size](io::ExpectedSize result) {
                        StreamNextWriteHandler(entry_size, result);
                }));
}

void UpdateModule::StreamOpenHandler(io::ExpectedAsyncWriterPtr writer) {
        if (!writer) {
                DownloadErrorHandler(writer.error());
                return;
        }
        download_->current_stream_writer_ = writer.value();

        DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(
                download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {
                        PayloadReadHandler(result);
                }));
}

void UpdateModule::StreamNextWriteHandler(size_t expected_n, io::ExpectedSize result) {
        // Close stream-next writer.
        download_->stream_next_writer_.reset();
        if (!result) {
                DownloadErrorHandler(result.error());
        } else if (expected_n != result.value()) {
                DownloadErrorHandler(error::Error(
                        make_error_condition(errc::io_error),
                        "Unexpected number of written bytes to stream-next"));
        }
}

void UpdateModule::PayloadReadHandler(io::ExpectedSize result) {
        if (!result) {
                // Close streams.
                download_->current_stream_writer_.reset();
                download_->current_payload_reader_.reset();
                DownloadErrorHandler(result.error());
        } else if (result.value() > 0) {
                DownloadErrorHandler(download_->current_stream_writer_->AsyncWrite(
                        download_->buffer_.begin(),
                        download_->buffer_.begin() + result.value(),
                        [this, result](io::ExpectedSize write_result) {
                                StreamWriteHandler(result.value(), write_result);
                        }));
        } else {
                // Close streams.
                download_->current_stream_writer_.reset();
                download_->current_payload_reader_.reset();

                if (download_->downloading_to_files_) {
                        StartDownloadToFile();
                } else {
                        DownloadErrorHandler(OpenStreamNextPipe(
                                [this](io::ExpectedAsyncWriterPtr writer) { StreamNextOpenHandler(writer); }));
                }
        }
}

void UpdateModule::StreamWriteHandler(size_t expected_n, io::ExpectedSize result) {
        if (!result) {
                DownloadErrorHandler(result.error());
        } else if (expected_n != result.value()) {
                DownloadErrorHandler(error::Error(
                        make_error_condition(errc::io_error),
                        "Unexpected number of written bytes to download stream"));
        } else {
                download_->written_ += result.value();
                log::Trace("Wrote " + to_string(download_->written_) + " bytes to Update Module");
                DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(
                        download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {
                                PayloadReadHandler(result);
                        }));
        }
}

void UpdateModule::EndStreamNext() {
        // Empty write.
        DownloadErrorHandler(download_->stream_next_writer_->AsyncWrite(
                download_->buffer_.begin(), download_->buffer_.begin(), [this](io::ExpectedSize result) {
                        if (!result) {
                                DownloadErrorHandler(result.error());
                        } else {
                                DownloadErrorHandler(error::NoError);
                        }
                        // Close writer.
                        download_->stream_next_writer_.reset();
                        // No further action necessary. Now we just need to wait for the process to finish.
                }));
}

void UpdateModule::DownloadErrorHandler(const error::Error &err) {
        if (err != error::NoError) {
                EndDownloadLoop(err);
        }
}

void UpdateModule::EndDownloadLoop(const error::Error &err) {
        download_->result_ = err;
        download_->event_loop_.Stop();
}

void UpdateModule::DownloadTimeoutHandler() {
        download_->proc_->EnsureTerminated();
        EndDownloadLoop(error::Error(
                make_error_condition(errc::timed_out), "Update Module Download process timed out"));
}

void UpdateModule::ProcessEndedHandler(error::Error err) {
        if (err != error::NoError) {
                err = GetProcessError(err);
                DownloadErrorHandler(error::Error(
                        err.code, "Download: Update Module returned non-zero status: " + err.message));
        } else if (download_->module_has_finished_download_) {
                EndDownloadLoop(error::NoError);
        } else if (download_->module_has_started_download_) {
                DownloadErrorHandler(error::Error(
                        make_error_condition(errc::broken_pipe),
                        "Update Module started downloading, but did not finish"));
        } else {
                download_->downloading_to_files_ = true;
                download_->stream_next_opener_.reset();
                download_->current_stream_opener_.reset();
                err = DeleteStreamsFiles();
                if (err != error::NoError) {
                        DownloadErrorHandler(err);
                } else {
                        StartDownloadToFile();
                }
        }
}

void UpdateModule::StartDownloadToFile() {
        auto reader = download_->payload_.Next();
        if (!reader) {
                if (reader.error().code
                        == artifact::parser_error::MakeError(
                                   artifact::parser_error::NoMorePayloadFilesError, "")
                                   .code) {
                        log::Debug("Downloaded all files to `files` directory.");
                        EndDownloadLoop(error::NoError);
                } else {
                        DownloadErrorHandler(reader.error());
                }
                return;
        }
        auto payload_reader = make_shared<artifact::Reader>(std::move(reader.value()));
        download_->current_payload_reader_ =
                make_shared<events::io::AsyncReaderFromReader>(download_->event_loop_, payload_reader);
        download_->current_payload_name_ = payload_reader->Name();

        auto stream_path = path::Join(update_module_workdir_, string("files"));
        auto err = PrepareDownloadDirectory(stream_path);
        if (err != error::NoError) {
                DownloadErrorHandler(err);
                return;
        }

        stream_path = path::Join(stream_path, download_->current_payload_name_);

        auto current_stream_writer =
                make_shared<events::io::AsyncFileDescriptorWriter>(download_->event_loop_);
        err = current_stream_writer->Open(stream_path);
        if (err != error::NoError) {
                DownloadErrorHandler(err);
                return;
        }
        download_->current_stream_writer_ = current_stream_writer;

        DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(
                download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {
                        PayloadReadHandler(result);
                }));
}

} // namespace v3
} // namespace update_module
} // namespace update
} // namespace mender

1	// Copyright 2023 Northern.tech AS
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	#include <mender-update/update_module/v3/update_module.hpp>
16
17	#include <common/events.hpp>
18	#include <common/events_io.hpp>
19	#include <common/log.hpp>
20	#include <common/path.hpp>
21
22	namespace mender {
23	namespace update {
24	namespace update_module {
25	namespace v3 {
26
27	namespace log = mender::common::log;
28	namespace path = mender::common::path;
29
30	void UpdateModule::StartDownloadProcess() {	40✔
31	log::Debug(	40✔
32	"Calling Update Module with command `" + update_module_path_ + " Download "	80✔
33	+ update_module_workdir_ + "`.");	120✔
34	download_->proc_ = make_shared<processes::Process>(	80✔
35	vector<string> {update_module_path_, "Download", update_module_workdir_});	240✔
36
37	download_->proc_->SetWorkDir(update_module_workdir_);	40✔
38
39	auto err = PrepareStreamNextPipe();	40✔
40	if (err != error::NoError) {	40✔
41	DownloadErrorHandler(err);	×
42	return;	×
43	}
44
45	err = download_->proc_->Start();	40✔
46	if (err != error::NoError) {	40✔
47	DownloadErrorHandler(GetProcessError(err));	×
48	return;	×
49	}
50
51	err = download_->proc_->AsyncWait(	80✔
52	download_->event_loop_, [this](error::Error err) { ProcessEndedHandler(err); });	118✔
53	if (err != error::NoError) {	40✔
54	DownloadErrorHandler(err);	×
55	return;	×
56	}
57
58	download_->proc_timeout_.AsyncWait(	80✔
59	chrono::seconds(ctx_.GetConfig().module_timeout_seconds), [this](error::Error err) {	40✔
60	if (err != error::NoError) {	1✔
61	DownloadErrorHandler(	×
62	err.WithContext("Error while waiting for Update Module Download process"));	×
63	} else {
64	DownloadTimeoutHandler();	1✔
65	}
66	});	81✔
67
68	DownloadErrorHandler(OpenStreamNextPipe(	40✔
69	[this](io::ExpectedAsyncWriterPtr writer) { StreamNextOpenHandler(writer); }));	87✔
70	}
71
72	void UpdateModule::StreamNextOpenHandler(io::ExpectedAsyncWriterPtr writer) {	10✔
73	if (!writer) {	10✔
74	DownloadErrorHandler(writer.error());	×
75	return;	2✔
76	}
77	download_->stream_next_writer_ = writer.value();	10✔
78
79	download_->module_has_started_download_ = true;	10✔
80
81	auto reader = download_->payload_.Next();	10✔
82	if (!reader) {	10✔
83	if (reader.error().code	4✔
84	== artifact::parser_error::MakeError(	2✔
85	artifact::parser_error::NoMorePayloadFilesError, "")	4✔
86	.code) {
87	download_->module_has_finished_download_ = true;	2✔
88	log::Debug("Update Module finished all downloads");	2✔
89	EndStreamNext();	2✔
90	} else {
91	DownloadErrorHandler(reader.error());	×
92	}
93	return;	2✔
94	}
95	auto payload_reader = make_shared<artifact::Reader>(std::move(reader.value()));	8✔
96	download_->current_payload_reader_ =	8✔
97	make_shared<events::io::AsyncReaderFromReader>(download_->event_loop_, payload_reader);	16✔
98	download_->current_payload_name_ = payload_reader->Name();	8✔
99
100	auto stream_path =
101	path::Join(update_module_workdir_, string("streams"), download_->current_payload_name_);	16✔
102	DownloadErrorHandler(PrepareAndOpenStreamPipe(	8✔
103	stream_path, [this](io::ExpectedAsyncWriterPtr writer) { StreamOpenHandler(writer); }));	20✔
104
105	string stream_next_string = path::Join("streams", download_->current_payload_name_);	16✔
106	size_t entry_size = stream_next_string.size() + 1;	8✔
107	if (entry_size > download_->buffer_.size()) {	8✔
108	DownloadErrorHandler(error::Error(	×
109	make_error_condition(errc::no_buffer_space), "Payload name is too large for buffer"));	×
110	return;	×
111	}
112	copy(stream_next_string.begin(), stream_next_string.end(), download_->buffer_.begin());	8✔
113	download_->buffer_[entry_size - 1] = '\n';	8✔
114	DownloadErrorHandler(download_->stream_next_writer_->AsyncWrite(	16✔
115	download_->buffer_.begin(),	8✔
116	download_->buffer_.begin() + entry_size,	8✔
117	[this, entry_size](io::ExpectedSize result) {	16✔
118	StreamNextWriteHandler(entry_size, result);	8✔
119	}));	40✔
120	}
121
122	void UpdateModule::StreamOpenHandler(io::ExpectedAsyncWriterPtr writer) {	4✔
123	if (!writer) {	4✔
124	DownloadErrorHandler(writer.error());	×
125	return;	×
126	}
127	download_->current_stream_writer_ = writer.value();	4✔
128
129	DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(	8✔
130	download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {	8✔
131	PayloadReadHandler(result);	4✔
132	}));	20✔
133	}
134
135	void UpdateModule::StreamNextWriteHandler(size_t expected_n, io::ExpectedSize result) {	8✔
136	// Close stream-next writer.
137	download_->stream_next_writer_.reset();	8✔
138	if (!result) {	8✔
139	DownloadErrorHandler(result.error());	1✔
140	} else if (expected_n != result.value()) {	7✔
141	DownloadErrorHandler(error::Error(	×
142	make_error_condition(errc::io_error),	×
143	"Unexpected number of written bytes to stream-next"));	×
144	}
145	}	8✔
146
147	void UpdateModule::PayloadReadHandler(io::ExpectedSize result) {	511✔
148	if (!result) {	511✔
149	// Close streams.
150	download_->current_stream_writer_.reset();	×
151	download_->current_payload_reader_.reset();	×
152	DownloadErrorHandler(result.error());	×
153	} else if (result.value() > 0) {	511✔
154	DownloadErrorHandler(download_->current_stream_writer_->AsyncWrite(	952✔
155	download_->buffer_.begin(),	476✔
156	download_->buffer_.begin() + result.value(),	476✔
157	[this, result](io::ExpectedSize write_result) {	476✔
158	StreamWriteHandler(result.value(), write_result);	476✔
159	}));	2,380✔
160	} else {
161	// Close streams.
162	download_->current_stream_writer_.reset();	35✔
163	download_->current_payload_reader_.reset();	35✔
164
165	if (download_->downloading_to_files_) {	35✔
166	StartDownloadToFile();	31✔
167	} else {
168	DownloadErrorHandler(OpenStreamNextPipe(	4✔
169	[this](io::ExpectedAsyncWriterPtr writer) { StreamNextOpenHandler(writer); }));	11✔
170	}
171	}
172	}	511✔
173
174	void UpdateModule::StreamWriteHandler(size_t expected_n, io::ExpectedSize result) {	476✔
175	if (!result) {	476✔
176	DownloadErrorHandler(result.error());	×
177	} else if (expected_n != result.value()) {	476✔
178	DownloadErrorHandler(error::Error(	×
179	make_error_condition(errc::io_error),	×
180	"Unexpected number of written bytes to download stream"));	×
181	} else {
182	download_->written_ += result.value();	476✔
183	log::Trace("Wrote " + to_string(download_->written_) + " bytes to Update Module");	476✔
184	DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(	952✔
185	download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {	952✔
186	PayloadReadHandler(result);	476✔
187	}));	2,380✔
188	}
189	}	476✔
190
191	void UpdateModule::EndStreamNext() {	2✔
192	// Empty write.
193	DownloadErrorHandler(download_->stream_next_writer_->AsyncWrite(	4✔
194	download_->buffer_.begin(), download_->buffer_.begin(), [this](io::ExpectedSize result) {	4✔
195	if (!result) {	2✔
196	DownloadErrorHandler(result.error());	×
197	} else {
198	DownloadErrorHandler(error::NoError);	2✔
199	}
200	// Close writer.
201	download_->stream_next_writer_.reset();	2✔
202	// No further action necessary. Now we just need to wait for the process to finish.
203	}));	8✔
204	}	2✔
205
206	void UpdateModule::DownloadErrorHandler(const error::Error &err) {	1,058✔
207	if (err != error::NoError) {	1,058✔
208	EndDownloadLoop(err);	7✔
209	}
210	}	1,058✔
211
212	void UpdateModule::EndDownloadLoop(const error::Error &err) {	40✔
213	download_->result_ = err;	40✔
214	download_->event_loop_.Stop();	40✔
215	}	40✔
216
217	void UpdateModule::DownloadTimeoutHandler() {	1✔
218	download_->proc_->EnsureTerminated();	1✔
219	EndDownloadLoop(error::Error(	1✔
220	make_error_condition(errc::timed_out), "Update Module Download process timed out"));	2✔
221	}	1✔
222
223	void UpdateModule::ProcessEndedHandler(error::Error err) {	38✔
224	if (err != error::NoError) {	38✔
225	err = GetProcessError(err);	3✔
226	DownloadErrorHandler(error::Error(	3✔
227	err.code, "Download: Update Module returned non-zero status: " + err.message));	6✔
228	} else if (download_->module_has_finished_download_) {	35✔
229	EndDownloadLoop(error::NoError);	2✔
230	} else if (download_->module_has_started_download_) {	33✔
231	DownloadErrorHandler(error::Error(	2✔
232	make_error_condition(errc::broken_pipe),	2✔
233	"Update Module started downloading, but did not finish"));	6✔
234	} else {
235	download_->downloading_to_files_ = true;	31✔
236	download_->stream_next_opener_.reset();	31✔
237	download_->current_stream_opener_.reset();	31✔
238	err = DeleteStreamsFiles();	31✔
239	if (err != error::NoError) {	31✔
240	DownloadErrorHandler(err);	×
241	} else {
242	StartDownloadToFile();	31✔
243	}
244	}
245	}	38✔
246
247	void UpdateModule::StartDownloadToFile() {	62✔
248	auto reader = download_->payload_.Next();	62✔
249	if (!reader) {	62✔
250	if (reader.error().code	60✔
251	== artifact::parser_error::MakeError(	30✔
252	artifact::parser_error::NoMorePayloadFilesError, "")	60✔
253	.code) {
254	log::Debug("Downloaded all files to `files` directory.");	30✔
255	EndDownloadLoop(error::NoError);	30✔
256	} else {
257	DownloadErrorHandler(reader.error());	×
258	}
259	return;	30✔
260	}
261	auto payload_reader = make_shared<artifact::Reader>(std::move(reader.value()));	32✔
262	download_->current_payload_reader_ =	32✔
263	make_shared<events::io::AsyncReaderFromReader>(download_->event_loop_, payload_reader);	64✔
264	download_->current_payload_name_ = payload_reader->Name();	32✔
265
266	auto stream_path = path::Join(update_module_workdir_, string("files"));	64✔
267	auto err = PrepareDownloadDirectory(stream_path);	32✔
268	if (err != error::NoError) {	32✔
269	DownloadErrorHandler(err);	×
270	return;	×
271	}
272
273	stream_path = path::Join(stream_path, download_->current_payload_name_);	32✔
274
275	auto current_stream_writer =
276	make_shared<events::io::AsyncFileDescriptorWriter>(download_->event_loop_);	32✔
277	err = current_stream_writer->Open(stream_path);	32✔
278	if (err != error::NoError) {	32✔
279	DownloadErrorHandler(err);	1✔
280	return;	1✔
281	}
282	download_->current_stream_writer_ = current_stream_writer;	31✔
283
284	DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(	62✔
285	download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {	62✔
286	PayloadReadHandler(result);	31✔
287	}));	155✔
288	}
289
290	} // namespace v3
291	} // namespace update_module
292	} // namespace update
293	} // namespace mender

mendersoftware / mender / 951615729

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