Implicitly extract tar.gz files from plugins that upload them

pkg/plugin/aggregation/aggregator.go

@@ -30,6 +30,7 @@ import (
 
  "github.com/golang/glog"
  "github.com/heptio/sonobuoy/pkg/plugin"
+ "github.com/viniciuschiele/tarx"
 )
 
 // Aggregator is responsible for taking results from an HTTP server (configured
@@ -196,26 +197,55 @@ func (a *Aggregator) handleResult(result *plugin.Result) error {
  // Create the output directory for the result. Will be of the
  // form .../plugins/:results_type/:node.json (for DaemonSet plugins) or
  // .../plugins/:results_type.json (for Job plugins)
- resultsFile := path.Join(a.OutputDir, result.Path()+result.Extension())
+ resultsFile := path.Join(a.OutputDir, result.Path()+result.Extension)
  resultsDir := path.Dir(resultsFile)
  glog.Infof("Creating directory %v", resultsDir)
  if err := os.MkdirAll(resultsDir, 0755); err != nil {
  glog.Errorf("Could not make directory %v: %v", resultsDir, err)
-
  return err
  }
 
- // Open the results file for writing
- f, err := os.Create(resultsFile)
+ // Write the results file out and close it
+ err := func() error {
+ f, err := os.Create(resultsFile)
+ if err != nil {
+ glog.Errorf("Could not open output file %v for writing: %v", resultsFile, err)
+ return err
+ }
+ defer f.Close()
+
+ // Copy the request body into the file
+ _, err = io.Copy(f, result.Body)
+ if err != nil {
+ glog.Errorf("Error writing plugin result: %v", err)
+ return err
+ }
+
+ return nil
+ }()
  if err != nil {
- glog.Errorf("Could not open output file %v for writing: %v", resultsFile, err)
  return err
  }
- defer f.Close()
 
- // Copy the request body into the file
- io.Copy(f, result.Body)
- glog.Infof("wrote results to %v\n", resultsFile)
+ // If it's a tarball, extract it
+ if result.Extension == ".tar.gz" {
+ resultsDir := path.Join(a.OutputDir, result.Path())
+
+ err = tarx.Extract(resultsFile, resultsDir, &tarx.ExtractOptions{})
+ if err != nil {
+ glog.Errorf("Could not extract tar file %v: %v", resultsFile, err)
+ return err
+ }
+
+ err = os.Remove(resultsFile)
+ if err != nil {
+ return err
+ }
+
+ glog.Infof("extracted results tarball into %v", resultsDir)
+ } else {
+ glog.Infof("wrote results to %v", resultsFile)
+ }
 
  return nil
 }

pkg/plugin/aggregation/aggregator_test.go

@@ -17,13 +17,17 @@ limitations under the License.
 package aggregation
 
 import (
+ "bytes"
  "io/ioutil"
  "os"
+ "os/exec"
  "path"
  "strconv"
  "testing"
 
  "github.com/heptio/sonobuoy/pkg/plugin"
+ pluginutils "github.com/heptio/sonobuoy/pkg/plugin/driver/utils"
+ "github.com/viniciuschiele/tarx"
 )
 
 func TestAggregation(t *testing.T) {
@@ -32,7 +36,7 @@ func TestAggregation(t *testing.T) {
  }
  // Happy path
  withAggregator(t, expected, func(agg *Aggregator) {
- resp := doRequest(t, "PUT", "/api/v1/results/by-node/node1/systemd_logs", "foo")
+ resp := doRequest(t, "PUT", "/api/v1/results/by-node/node1/systemd_logs.json", []byte("foo"))
  if resp.StatusCode != 200 {
  body, _ := ioutil.ReadAll(resp.Body)
  t.Errorf("Got (%v) response from server: %v", resp.StatusCode, string(body))
@@ -44,7 +48,59 @@ func TestAggregation(t *testing.T) {
  t.Errorf("results for node1 incorrect (got %v): %v", string(bytes), err)
  }
  } else {
- t.Errorf("AggregationServer didn't record a result for node1")
+ t.Errorf("AggregationServer didn't record a result for node1. Got: %+v", agg.Results)
+ }
+ })
+}
+
+func TestAggregation_noExtension(t *testing.T) {
+ expected := []plugin.ExpectedResult{
+ plugin.ExpectedResult{NodeName: "node1", ResultType: "systemd_logs"},
+ }
+
+ withAggregator(t, expected, func(agg *Aggregator) {
+ resp := doRequest(t, "PUT", "/api/v1/results/by-node/node1/systemd_logs", []byte("foo"))
+ if resp.StatusCode != 200 {
+ body, _ := ioutil.ReadAll(resp.Body)
+ t.Errorf("Got (%v) response from server: %v", resp.StatusCode, string(body))
+ }
+
+ if result, ok := agg.Results["systemd_logs/node1"]; ok {
+ bytes, err := ioutil.ReadFile(path.Join(agg.OutputDir, result.Path()))
+ if string(bytes) != "foo" {
+ t.Errorf("results for node1 incorrect (got %v): %v", string(bytes), err)
+ }
+ } else {
+ t.Errorf("AggregationServer didn't record a result for node1. Got: %+v", agg.Results)
+ }
+ })
+}
+
+func TestAggregation_tarfile(t *testing.T) {
+ expected := []plugin.ExpectedResult{
+ plugin.ExpectedResult{ResultType: "e2e"},
+ }
+
+ fileBytes := []byte("foo")
+ tarBytes := makeTarWithContents(t, "inside_tar.txt", fileBytes)
+
+ withAggregator(t, expected, func(agg *Aggregator) {
+ resp := doRequest(t, "PUT", "/api/v1/results/global/e2e.tar.gz", tarBytes)
+ if resp.StatusCode != 200 {
+ body, _ := ioutil.ReadAll(resp.Body)
+ t.Errorf("Got (%v) response from server: %v", resp.StatusCode, string(body))
+ }
+
+ if result, ok := agg.Results["e2e"]; ok {
+ realBytes, err := ioutil.ReadFile(path.Join(agg.OutputDir, result.Path(), "inside_tar.txt"))
+ if bytes.Compare(realBytes, fileBytes) != 0 || err != nil {
+ t.Logf("results e2e tests incorrect (got %v, expected %v): %v", string(realBytes), string(fileBytes), err)
+ output, _ := exec.Command("ls", "-lR", agg.OutputDir).CombinedOutput()
+ t.Log(string(output))
+ t.Fail()
+ }
+ } else {
+ t.Errorf("AggregationServer didn't record a result for e2e tests. Got: %+v", agg.Results)
  }
  })
 }
@@ -55,14 +111,13 @@ func TestAggregation_wrongnodes(t *testing.T) {
  }
 
  withAggregator(t, expected, func(agg *Aggregator) {
- resp := doRequest(t, "PUT", "/api/v1/results/by-node/node10/systemd_logs", "foo")
+ resp := doRequest(t, "PUT", "/api/v1/results/by-node/node10/systemd_logs.json", []byte("foo"))
  if resp.StatusCode != 403 {
  t.Errorf("Expected a 403 forbidden for checking in an unexpected node, got %v", resp.StatusCode)
  }
 
- if _, ok := agg.Results["node10"]; ok {
+ if _, ok := agg.Results["systemd_logs/node10"]; ok {
  t.Fatal("Aggregator accepted a result from an unexpected host")
- t.Fail()
  }
  })
 }
@@ -74,20 +129,43 @@ func TestAggregation_duplicates(t *testing.T) {
  }
  withAggregator(t, expected, func(agg *Aggregator) {
  // Check in a node
- resp := doRequest(t, "PUT", "/api/v1/results/by-node/node1/systemd_logs", "foo")
+ resp := doRequest(t, "PUT", "/api/v1/results/by-node/node1/systemd_logs.json", []byte("foo"))
  if resp.StatusCode != 200 {
  t.Errorf("Got non-200 response from server: %v", resp.StatusCode)
  }
 
  // Check in the same node again, should conflict
- resp = doRequest(t, "PUT", "/api/v1/results/by-node/node1/systemd_logs", "foo")
+ resp = doRequest(t, "PUT", "/api/v1/results/by-node/node1/systemd_logs.json", []byte("foo"))
  if resp.StatusCode != 409 {
  t.Errorf("Expected a 409 conflict for checking in a duplicate node, got %v", resp.StatusCode)
  }
 
  if _, ok := agg.Results["node10"]; ok {
  t.Fatal("Aggregator accepted a result from an unexpected host")
- t.Fail()
+ }
+ })
+}
+
+func TestAggregation_errors(t *testing.T) {
+ expected := []plugin.ExpectedResult{
+ plugin.ExpectedResult{ResultType: "e2e"},
+ }
+
+ withAggregator(t, expected, func(agg *Aggregator) {
+ resultsCh := make(chan *plugin.Result)
+ go agg.IngestResults(resultsCh)
+
+ // Send an error
+ resultsCh <- pluginutils.MakeErrorResult("e2e", map[string]interface{}{"error": "foo"}, "")
+ agg.Wait(make(chan bool))
+
+ if result, ok := agg.Results["e2e"]; ok {
+ bytes, err := ioutil.ReadFile(path.Join(agg.OutputDir, result.Path()) + ".json")
+ if err != nil || string(bytes) != `{"error":"foo"}` {
+ t.Errorf("results for e2e plugin incorrect (got %v): %v", string(bytes), err)
+ }
+ } else {
+ t.Errorf("Aggregator didn't record error result from e2e plugin, got %v", agg.Results)
  }
  })
 }
@@ -117,3 +195,44 @@ func withAggregator(t *testing.T, expected []plugin.ExpectedResult, callback fun
  srv.WaitUntilReady()
  callback(agg)
 }
+
+// Create a gzipped tar file with the given filename (and contents) inside it,
+// return the raw bytes for that tar file.
+func makeTarWithContents(t *testing.T, filename string, fileContents []byte) (tarbytes []byte) {
+ dir, err := ioutil.TempDir("", "sonobuoy_server_test")
+ if err != nil {
+ t.Fatalf("Could not create temp directory: %v", err)
+ return
+ }
+ defer os.RemoveAll(dir)
+
+ tardir := path.Join(dir, "results")
+ err = os.Mkdir(tardir, 0755)
+ if err != nil {
+ t.Fatal("Could not create results directory %v: %v", tardir, err)
+ return
+ }
+
+ filepath := path.Join(tardir, filename)
+ tarfile := path.Join(dir, "results.tar.gz")
+
+ err = ioutil.WriteFile(filepath, fileContents, 0644)
+ if err != nil {
+ t.Fatalf("Could not write to temp file %v: %v", filepath, err)
+ return
+ }
+
+ err = tarx.Compress(tarfile, tardir, &tarx.CompressOptions{Compression: tarx.Gzip})
+ if err != nil {
+ t.Fatalf("Could not create tar file %v: %v", tarfile, err)
+ return
+ }
+
+ tarbytes, err = ioutil.ReadFile(tarfile)
+ if err != nil {
+ t.Fatalf("Could not read created tar file %v: %v", tarfile, err)
+ return
+ }
+
+ return tarbytes
+}

pkg/plugin/aggregation/server.go

@@ -153,13 +153,15 @@ func (s *Server) nodeResultsHandler(w http.ResponseWriter, r *http.Request) {
  return
  }
 
- // Parse the path into the node name and the type
- node, resultType := parts[0], parts[1]
+ // Parse the path into the node name, result type, and extension
+ node, file := parts[0], parts[1]
+ resultType, extension := parseFileName(file)
 
  glog.Infof("got %v result from %v\n", resultType, node)
 
  result := &plugin.Result{
  ResultType: resultType,
+ Extension: extension,
  NodeName: node,
  Body: r.Body,
  }
@@ -183,7 +185,6 @@ func (s *Server) globalResultsHandler(w http.ResponseWriter, r *http.Request) {
  http.NotFound(w, r)
  return
  }
- resultType := parts[0]
 
  // We accept PUT because the client is specifying the resource identifier via
  // the HTTP path. (As opposed to POST, where typically the clients would post
@@ -198,11 +199,13 @@ func (s *Server) globalResultsHandler(w http.ResponseWriter, r *http.Request) {
  return
  }
 
+ resultType, extension := parseFileName(parts[0])
  glog.Infof("got %v result\n", resultType)
 
  result := &plugin.Result{
  NodeName: "",
  ResultType: resultType,
+ Extension: extension,
  Body: r.Body,
  }
 
@@ -212,3 +215,16 @@ func (s *Server) globalResultsHandler(w http.ResponseWriter, r *http.Request) {
  s.ResultsCallback(result, w)
  r.Body.Close()
 }
+
+// given an uploaded filename, parse it into its base name and extension. If
+// there are no "." characters, the extension will be blank and the name will
+// be set to the filename as-is
+func parseFileName(file string) (name string, extension string) {
+ filenameParts := strings.SplitN(file, ".", 2)
+
+ if len(filenameParts) == 2 {
+ return filenameParts[0], "." + filenameParts[1]
+ }
+
+ return file, ""
+}

pkg/plugin/aggregation/server_test.go

@@ -17,12 +17,12 @@ limitations under the License.
 package aggregation
 
 import (
+ "bytes"
  "io/ioutil"
  "net/http"
  "net/url"
  "os"
  "strconv"
- "strings"
  "testing"
 
  "github.com/heptio/sonobuoy/pkg/plugin"
@@ -32,7 +32,7 @@ func TestStart(t *testing.T) {
  checkins := make(map[string]*plugin.Result, 0)
 
  expectedResult := "systemd_logs/results/testnode"
- expectedJSON := `{"some": "json"}`
+ expectedJSON := []byte(`{"some": "json"}`)
 
  tmpdir, err := ioutil.TempDir("", "sonobuoy_server_test")
  if err != nil {
@@ -100,7 +100,7 @@ func TestStart(t *testing.T) {
 
 var testPort = 8099
 
-func doRequest(t *testing.T, method, path, body string) *http.Response {
+func doRequest(t *testing.T, method, path string, body []byte) *http.Response {
  // Make a new HTTP transport for every request, this avoids issues where HTTP
  // connection keep-alive leaves connections running to old server instances.
  // (We can take the performance hit since it's just tests.)
@@ -115,7 +115,7 @@ func doRequest(t *testing.T, method, path, body string) *http.Response {
  req, err := http.NewRequest(
  method,
  resultsURL.String(),
- strings.NewReader(body),
+ bytes.NewReader(body),
  )
  if err != nil {
  t.Fatalf("error constructing request: %v", err)

pkg/plugin/driver/daemonset/daemonset.go

@@ -213,7 +213,7 @@ func (p *Plugin) Monitor(kubeclient kubernetes.Interface, availableNodes []v1.No
  if isFailing, reason := utils.IsPodFailing(&pod); isFailing {
  podsReported[nodeName] = true
 
- resultsCh <- utils.MakeErrorResult(p, map[string]interface{}{
+ resultsCh <- utils.MakeErrorResult(p.GetResultType(), map[string]interface{}{
  "error": reason,
  "pod": pod,
  }, nodeName)
@@ -228,7 +228,7 @@ func (p *Plugin) Monitor(kubeclient kubernetes.Interface, availableNodes []v1.No
  for _, node := range availableNodes {
  if !podsFound[node.Name] && !podsReported[node.Name] {
  podsReported[node.Name] = true
- resultsCh <- utils.MakeErrorResult(p, map[string]interface{}{
+ resultsCh <- utils.MakeErrorResult(p.GetResultType(), map[string]interface{}{
  "error": fmt.Sprintf(
  "No pod was scheduled on node %v within %v. Check tolerations for plugin %v",
  node.Name,