Added improved error handling around downloading and installing packages

CHANGELOG.md

@@ -7,16 +7,19 @@
   removed on startup ([#312]).
 
 ### Changed
+- Changed the version reported by the Stackable Agent in `nodeInfo.kubeletVersion` of the `Node` object in Kubernetes
+  from the version of the Krustlet library to the Stackable Agent version ([#315]).
 - Restart agent on all crashes ([#318]).
+- Agent will now request content type "application/gzip" in package downloads and reject responses with content type
+  that is not one of either "application/gzip", "application/tgz" or "application/x-gzip" ([#326])
 
-[#312]: https://github.com/stackabletech/agent/pull/312
-[#318]: https://github.com/stackabletech/agent/pull/318
-
-### Changed
-- Changed the version reported by the Stackable Agent in `nodeInfo.kubeletVersion` of the `Node` object in Kubernetes 
-  from the version of the Krustlet library to the Stackable Agent version ([#315]).
+### Fixed
+- Agent deletes directories from failed install attempts ([#326])
 
+[#312]: https://github.com/stackabletech/agent/pull/312
 [#315]: https://github.com/stackabletech/agent/pull/315
+[#318]: https://github.com/stackabletech/agent/pull/318
+[#326]: https://github.com/stackabletech/agent/pull/326
 
 ## [0.6.1] - 2021-09-14
 

src/provider/error.rs

@@ -3,6 +3,7 @@ use k8s_openapi::url;
 use thiserror::Error;
 
 use crate::provider::repository::package::Package;
+use reqwest::Url;
 use std::ffi::OsString;
 
 #[derive(Error, Debug)]
@@ -22,6 +23,12 @@ pub enum StackableError {
         #[from]
         source: kube::Error,
     },
+    #[error("An error has ocurred when trying to download [{package}] from [{download_link}]: {errormessage}")]
+    PackageDownloadError {
+        package: Package,
+        download_link: Url,
+        errormessage: String,
+    },
     #[error(transparent)]
     TemplateRenderError(#[from] RenderError),
     #[error(transparent)]

src/provider/repository/stackablerepository.rs

@@ -3,18 +3,26 @@ use std::convert::TryFrom;
 use std::fmt;
 use std::fs::File;
 use std::hash::{Hash, Hasher};
-use std::io::{copy, Cursor};
+use std::io::{copy, Cursor, Write};
 use std::path::PathBuf;
 
+use crate::provider::error::StackableError;
+use crate::provider::error::StackableError::{PackageDownloadError, PackageNotFound};
+use crate::provider::repository::package::Package;
+use crate::provider::repository::repository_spec::Repository;
 use kube::api::Meta;
-use log::{debug, trace};
+use log::{debug, trace, warn};
+use reqwest::header::{ACCEPT, CONTENT_TYPE};
+use reqwest::{Client, StatusCode};
 use serde::{Deserialize, Serialize};
 use url::Url;
 
-use crate::provider::error::StackableError;
-use crate::provider::error::StackableError::PackageNotFound;
-use crate::provider::repository::package::Package;
-use crate::provider::repository::repository_spec::Repository;
+// These are the default content_types that we have seen in the wild
+// of these only 'application/gzip' is valid according to
+// https://www.iana.org/assignments/media-types/media-types.xhtml but our own
+// Nexus uses the other two, so we cannot really complain
+const DEFAULT_ALLOWED_CONTENT_TYPES: &[&str] =
+    &["application/gzip", "application/tgz", "application/x-gzip"];
 
 #[derive(Debug, Clone)]
 pub struct StackableRepoProvider {
@@ -112,12 +120,83 @@ impl StackableRepoProvider {
 
         let stackable_package = self.get_package(package.clone()).await?;
         let download_link = Url::parse(&stackable_package.link)?;
-        let response = reqwest::get(download_link).await?;
 
+        let client = Client::builder()
+            .build()
+            .map_err(|error| PackageDownloadError {
+                package: package.clone(),
+                download_link: download_link.clone(),
+                errormessage: format!("Unable to create http client: [{}]", error),
+            })?;
+
+        // We set the ACCEPT header field on our request which states that the only content type
+        // we are willing to accept is 'application/gzip'
+        // If the webserver is unable to provide this content type to us it _SHOULD_ respond with a
+        // 406 response code, but it seems we can't rely on that.
+        // For more details see: https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.1
+        let response = match client
+            .get(download_link.clone())
+            .header(ACCEPT, "application/gzip")
+            .send()
+            .await
+        {
+            Ok(response) if response.status().is_success() => {
+                // The request was successful, but just to be safe we'll still check the content_type, 
+                // since the webserver is free to ignore the requested content_type
+                if let Some(content_type) = response.headers().get(CONTENT_TYPE) {
+                    let content_type = content_type.to_str().map_err(|error| PackageDownloadError {
+                        package: package.clone(),
+                        download_link: download_link.clone(),
+                        errormessage: format!("Got content_type with non-ascii characters from webserver: [{}]", error),
+                    })?;
+
+                    if DEFAULT_ALLOWED_CONTENT_TYPES.contains(&content_type) {
+                        Ok(response)
+                    } else {
+                        // If we get a known wrong content type we'll abort
+                        Err(PackageDownloadError {
+                            package: package.clone(),
+                            download_link,
+                            errormessage: format!(
+                                "Got wrong 'content_type' header [{:?}] in response from webserver.",
+                                content_type
+                            ),
+                        })
+                    }
+                } else {
+                    // If we get no content_type (not sure if this is even legal) we'll soldier on and hope for the best
+                    debug!("Response had no 'content_type' header set, we'll give the sender the benefit of the doubt and try processing this anyway.");
+                    Ok(response)
+                }
+            }
+            Ok(response) if response.status() == StatusCode::NOT_ACCEPTABLE =>
+                Err(PackageDownloadError {
+                    package: package.clone(),
+                    download_link,
+                    errormessage: "Got response code 406 from webserver - Unable to negotiate content type, this is probably due to content encoding settings on the webserver.".to_string(),
+                }),
+            Ok(response) => Err(PackageDownloadError {
+                package: package.clone(),
+                download_link,
+                errormessage: format!(
+                    "Got non-success response [{}] from webserver!",
+                    response.status()
+                ),
+            }),
+            Err(error) => Err(PackageDownloadError {
+                package: package.clone(),
+                download_link,
+                errormessage: format!("{}", error),
+            }),
+        }?;
+
+        // All error cases return above, so we can safely assume that this is a valid download at
+        // this point
         let mut content = Cursor::new(response.bytes().await?);
 
         let mut out = File::create(target_path.join(package.get_file_name()))?;
         copy(&mut content, &mut out)?;
+        out.flush()?;
         Ok(())
     }
 
@@ -126,10 +205,26 @@ impl StackableRepoProvider {
 
         debug!("Retrieving repository metadata from {}", self.metadata_url);
 
-        let repo_data = reqwest::get(self.metadata_url.clone())
-            .await?
-            .json::<RepoData>()
-            .await?;
+        let repo_data = match reqwest::get(self.metadata_url.clone()).await {
+            Ok(repo_data) => repo_data,
+            Err(error) => {
+                warn!(
+                    "Failed to retrieve metadata from {} due to {:?}",
+                    self.metadata_url, error
+                );
+                return Err(error.into());
+            }
+        };
+        let repo_data = match repo_data.json::<RepoData>().await {
+            Ok(parsed_data) => parsed_data,
+            Err(error) => {
+                warn!(
+                    "Error parsing metadata from repository {}: {:?}",
+                    self.name, error
+                );
+                return Err(error.into());
+            }
+        };
 
         debug!("Got repository metadata: {:?}", repo_data);
 

src/provider/states/pod/installing.rs

@@ -1,5 +1,6 @@
+use std::fs;
 use std::fs::File;
-use std::path::{Path, PathBuf};
+use std::path::PathBuf;
 
 use flate2::read::GzDecoder;
 use kubelet::pod::state::prelude::*;
@@ -25,15 +26,15 @@ impl Installing {
     fn package_installed<T: Into<Package>>(&self, package: T) -> bool {
         let package = package.into();
 
-        let package_file_name = self.parcel_directory.join(package.get_directory_name());
+        let target_directory = self.get_target_directory(&package);
         debug!(
             "Checking if package {:?} has already been installed to {:?}",
-            package, package_file_name
+            package, target_directory
         );
-        Path::new(&package_file_name).exists()
+        target_directory.exists()
     }
 
-    fn get_target_directory(&self, package: Package) -> PathBuf {
+    fn get_target_directory(&self, package: &Package) -> PathBuf {
         self.parcel_directory.join(package.get_directory_name())
     }
 
@@ -45,13 +46,13 @@ impl Installing {
         let tar = GzDecoder::new(tar_gz);
         let mut archive = Archive::new(tar);
 
-        let target_directory = self.get_target_directory(package.clone());
+        let target_directory = self.get_target_directory(&package);
 
         info!(
             "Installing package: {:?} from {:?} into {:?}",
             package, archive_path, target_directory
         );
-        archive.unpack(self.parcel_directory.join(package.get_directory_name()))?;
+        archive.unpack(target_directory)?;
         Ok(())
     }
 }
@@ -76,7 +77,7 @@ impl State<PodState> for Installing {
             );
         } else {
             info!("Installing package {}", package);
-            match self.install_package(package) {
+            match self.install_package(package.clone()) {
                 Ok(()) => Transition::next(
                     self,
                     CreatingConfig {
@@ -88,6 +89,20 @@ impl State<PodState> for Installing {
                         "Failed to install package [{}] due to: [{:?}]",
                         &package_name, e
                     );
+                    // Clean up partially unpacked directory to avoid later iterations assuming
+                    // this install attempt was successful because the target directory exists.
+                    let installation_directory = self.get_target_directory(&package);
+                    debug!(
+                        "Cleaning up partial installation by deleting directory [{}]",
+                        installation_directory.to_string_lossy()
+                    );
+                    if let Err(error) = fs::remove_dir_all(&installation_directory) {
+                        error!(
+                            "Failed to clean up directory [{}] due to {}",
+                            installation_directory.to_string_lossy(),
+                            error
+                        );
+                    };
                     Transition::next(
                         self,
                         SetupFailed {