950534094

pending completion

Build # 950534094

Build Type

push

gitlab-ci

Committed by

kacf

Commit Message

chore: Add `StartsWith` and `EndsWith` generic utility functions.

Also use the latter in `states.cpp`.

Signed-off-by: Kristian Amlie <kristian.amlie@northern.tech>

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

4931 of 6276 relevant lines covered (78.57%)

196.18 hits per line

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

84.49

/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<procs::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_->download_finished_handler_(err);
}

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() {	60✔
31	log::Debug(	60✔
32	"Calling Update Module with command `" + update_module_path_ + " Download "	120✔
33	+ update_module_workdir_ + "`.");	180✔
34	download_->proc_ = make_shared<procs::Process>(	120✔
35	vector<string> {update_module_path_, "Download", update_module_workdir_});	360✔
36
37	download_->proc_->SetWorkDir(update_module_workdir_);	60✔
38
39	auto err = PrepareStreamNextPipe();	60✔
40	if (err != error::NoError) {	60✔
41	DownloadErrorHandler(err);	×
42	return;	×
43	}
44
45	err = download_->proc_->Start();	60✔
46	if (err != error::NoError) {	60✔
47	DownloadErrorHandler(GetProcessError(err));	×
48	return;	×
49	}
50
51	err = download_->proc_->AsyncWait(	120✔
52	download_->event_loop_, [this](error::Error err) { ProcessEndedHandler(err); });	176✔
53	if (err != error::NoError) {	60✔
54	DownloadErrorHandler(err);	×
55	return;	×
56	}
57
58	download_->proc_timeout_.AsyncWait(	120✔
59	chrono::seconds(ctx_.GetConfig().module_timeout_seconds), [this](error::Error err) {	60✔
60	if (err != error::NoError) {	2✔
61	DownloadErrorHandler(	×
62	err.WithContext("Error while waiting for Update Module Download process"));	×
63	} else {
64	DownloadTimeoutHandler();	2✔
65	}
66	});	122✔
67
68	DownloadErrorHandler(OpenStreamNextPipe(	60✔
69	[this](io::ExpectedAsyncWriterPtr writer) { StreamNextOpenHandler(writer); }));	128✔
70	}
71
72	void UpdateModule::StreamNextOpenHandler(io::ExpectedAsyncWriterPtr writer) {	11✔
73	if (!writer) {	11✔
74	DownloadErrorHandler(writer.error());	×
75	return;	2✔
76	}
77	download_->stream_next_writer_ = writer.value();	11✔
78
79	download_->module_has_started_download_ = true;	11✔
80
81	auto reader = download_->payload_.Next();	11✔
82	if (!reader) {	11✔
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()));	9✔
96	download_->current_payload_reader_ =	9✔
97	make_shared<events::io::AsyncReaderFromReader>(download_->event_loop_, payload_reader);	18✔
98	download_->current_payload_name_ = payload_reader->Name();	9✔
99
100	auto stream_path =
101	path::Join(update_module_workdir_, string("streams"), download_->current_payload_name_);	18✔
102	DownloadErrorHandler(PrepareAndOpenStreamPipe(	9✔
103	stream_path, [this](io::ExpectedAsyncWriterPtr writer) { StreamOpenHandler(writer); }));	23✔
104
105	string stream_next_string = path::Join("streams", download_->current_payload_name_);	18✔
106	size_t entry_size = stream_next_string.size() + 1;	9✔
107	if (entry_size > download_->buffer_.size()) {	9✔
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());	9✔
113	download_->buffer_[entry_size - 1] = '\n';	9✔
114	DownloadErrorHandler(download_->stream_next_writer_->AsyncWrite(	18✔
115	download_->buffer_.begin(),	9✔
116	download_->buffer_.begin() + entry_size,	9✔
117	[this, entry_size](io::ExpectedSize result) {	18✔
118	StreamNextWriteHandler(entry_size, result);	9✔
119	}));	45✔
120	}
121
122	void UpdateModule::StreamOpenHandler(io::ExpectedAsyncWriterPtr writer) {	5✔
123	if (!writer) {	5✔
124	DownloadErrorHandler(writer.error());	×
125	return;	×
126	}
127	download_->current_stream_writer_ = writer.value();	5✔
128
129	DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(	10✔
130	download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {	10✔
131	PayloadReadHandler(result);	5✔
132	}));	25✔
133	}
134
135	void UpdateModule::StreamNextWriteHandler(size_t expected_n, io::ExpectedSize result) {	9✔
136	// Close stream-next writer.
137	download_->stream_next_writer_.reset();	9✔
138	if (!result) {	9✔
139	DownloadErrorHandler(result.error());	1✔
140	} else if (expected_n != result.value()) {	8✔
141	DownloadErrorHandler(error::Error(	×
142	make_error_condition(errc::io_error),	×
143	"Unexpected number of written bytes to stream-next"));	×
144	}
145	}	9✔
146
147	void UpdateModule::PayloadReadHandler(io::ExpectedSize result) {	513✔
148	if (!result) {	513✔
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) {	513✔
154	DownloadErrorHandler(download_->current_stream_writer_->AsyncWrite(	956✔
155	download_->buffer_.begin(),	478✔
156	download_->buffer_.begin() + result.value(),	478✔
157	[this, result](io::ExpectedSize write_result) {	478✔
158	StreamWriteHandler(result.value(), write_result);	478✔
159	}));	2,390✔
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	}	513✔
173
174	void UpdateModule::StreamWriteHandler(size_t expected_n, io::ExpectedSize result) {	478✔
175	if (!result) {	478✔
176	DownloadErrorHandler(result.error());	1✔
177	} else if (expected_n != result.value()) {	477✔
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();	477✔
183	log::Trace("Wrote " + to_string(download_->written_) + " bytes to Update Module");	477✔
184	DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(	954✔
185	download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {	954✔
186	PayloadReadHandler(result);	477✔
187	}));	2,385✔
188	}
189	}	478✔
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,086✔
207	if (err != error::NoError) {	1,086✔
208	EndDownloadLoop(err);	9✔
209	}
210	}	1,086✔
211
212	void UpdateModule::EndDownloadLoop(const error::Error &err) {	60✔
213	download_->download_finished_handler_(err);	60✔
214	}	60✔
215
216	void UpdateModule::DownloadTimeoutHandler() {	2✔
217	download_->proc_->EnsureTerminated();	2✔
218	EndDownloadLoop(error::Error(	2✔
219	make_error_condition(errc::timed_out), "Update Module Download process timed out"));	4✔
220	}	2✔
221
222	void UpdateModule::ProcessEndedHandler(error::Error err) {	56✔
223	if (err != error::NoError) {	56✔
224	err = GetProcessError(err);	4✔
225	DownloadErrorHandler(error::Error(	4✔
226	err.code, "Download: Update Module returned non-zero status: " + err.message));	8✔
227	} else if (download_->module_has_finished_download_) {	52✔
228	EndDownloadLoop(error::NoError);	2✔
229	} else if (download_->module_has_started_download_) {	50✔
230	DownloadErrorHandler(error::Error(	2✔
231	make_error_condition(errc::broken_pipe),	2✔
232	"Update Module started downloading, but did not finish"));	6✔
233	} else {
234	download_->downloading_to_files_ = true;	48✔
235	download_->stream_next_opener_.reset();	48✔
236	download_->current_stream_opener_.reset();	48✔
237	err = DeleteStreamsFiles();	48✔
238	if (err != error::NoError) {	48✔
239	DownloadErrorHandler(err);	×
240	} else {
241	StartDownloadToFile();	48✔
242	}
243	}
244	}	56✔
245
246	void UpdateModule::StartDownloadToFile() {	79✔
247	auto reader = download_->payload_.Next();	79✔
248	if (!reader) {	79✔
249	if (reader.error().code	94✔
250	== artifact::parser_error::MakeError(	47✔
251	artifact::parser_error::NoMorePayloadFilesError, "")	94✔
252	.code) {
253	log::Debug("Downloaded all files to `files` directory.");	47✔
254	EndDownloadLoop(error::NoError);	47✔
255	} else {
256	DownloadErrorHandler(reader.error());	×
257	}
258	return;	47✔
259	}
260	auto payload_reader = make_shared<artifact::Reader>(std::move(reader.value()));	32✔
261	download_->current_payload_reader_ =	32✔
262	make_shared<events::io::AsyncReaderFromReader>(download_->event_loop_, payload_reader);	64✔
263	download_->current_payload_name_ = payload_reader->Name();	32✔
264
265	auto stream_path = path::Join(update_module_workdir_, string("files"));	64✔
266	auto err = PrepareDownloadDirectory(stream_path);	32✔
267	if (err != error::NoError) {	32✔
268	DownloadErrorHandler(err);	×
269	return;	×
270	}
271
272	stream_path = path::Join(stream_path, download_->current_payload_name_);	32✔
273
274	auto current_stream_writer =
275	make_shared<events::io::AsyncFileDescriptorWriter>(download_->event_loop_);	32✔
276	err = current_stream_writer->Open(stream_path);	32✔
277	if (err != error::NoError) {	32✔
278	DownloadErrorHandler(err);	1✔
279	return;	1✔
280	}
281	download_->current_stream_writer_ = current_stream_writer;	31✔
282
283	DownloadErrorHandler(download_->current_payload_reader_->AsyncRead(	62✔
284	download_->buffer_.begin(), download_->buffer_.end(), [this](io::ExpectedSize result) {	62✔
285	PayloadReadHandler(result);	31✔
286	}));	155✔
287	}
288
289	} // namespace v3
290	} // namespace update_module
291	} // namespace update
292	} // namespace mender

mendersoftware / mender / 950534094

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