Implicitly extract tar.gz files from plugins that upload them

This makes it so plugins upload to a path like:

```
/api/v1/results/global/e2e.tar.gz
```

And it will show up in:

```
/results/plugins/e2e/results/<contents>
```

(For single-node plugins, this would map to:)

```
/results/plugins/systemd_logs/node1/<contents>
```

This needs to be documented along with the snapshot documentation that
is already under review.

Since the upload path is now expected to have a file extension in it, In
case for some reason the sonobuoy master launches with an old version
of the sonobuoy worker (or for plugins not using `sonobuoy worker` that
upload their own contents), if the path has no file extension, it is
guessed to be `.json` or `.tar.gz`, same as before.

Signed-off-by: Ken Simon <ninkendo@gmail.com>

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
@@ -193,15 +194,22 @@ func (a *Aggregator) handleResult(result *plugin.Result) error {
  a.resultEvents <- result
  }()
 
+ // Get the file extension to save the result as, falling back on guessing it.
+ var extension string
+ if result.Extension != "" {
+ extension = result.Extension
+ } else {
+ extension = result.GuessExtension()
+ }
+
  // 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()+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
  }
 
@@ -217,5 +225,20 @@ func (a *Aggregator) handleResult(result *plugin.Result) error {
  io.Copy(f, result.Body)
  glog.Infof("wrote results to %v\n", resultsFile)
 
+ // If it's a tarball, extract it
+ if extension == ".tar.gz" {
+ f.Close()
+ err = tarx.Extract(resultsFile, path.Join(a.OutputDir, result.Path()), &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
+ }
+ }
+
  return nil
 }

pkg/plugin/aggregation/aggregator_test.go

@@ -17,13 +17,16 @@ limitations under the License.
 package aggregation
 
 import (
+ "bytes"
  "io/ioutil"
  "os"
+ "os/exec"
  "path"
  "strconv"
  "testing"
 
  "github.com/heptio/sonobuoy/pkg/plugin"
+ "github.com/viniciuschiele/tarx"
 )
 
 func TestAggregation(t *testing.T) {
@@ -32,7 +35,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 +47,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_guessExtension(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()) + ".json")
+ 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,7 +110,7 @@ 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)
  }
@@ -74,13 +129,13 @@ 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)
  }
@@ -117,3 +172,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/interface.go

@@ -77,6 +77,7 @@ type ExpectedResult struct {
 type Result struct {
  NodeName string
  ResultType string
+ Extension string
  Body io.Reader
  Error string
 }
@@ -145,9 +146,10 @@ func (r *Result) ExpectedResultID() string {
  return r.ResultType + "/" + r.NodeName
 }
 
-// Extension returns the results extension for different plugins
+// GuessExtension is for backwards compatibility with plugins that don't upload
+// with a file extension directly in the URL.
 // TODO: We should load this for the plugin.
-func (r *Result) Extension() string {
+func (r *Result) GuessExtension() string {
  if r.ResultType == "e2e" {
  return ".tar.gz"
  }

pkg/worker/worker.go

@@ -20,6 +20,7 @@ import (
  "io"
  "io/ioutil"
  "os"
+ "strings"
  "time"
 
  "github.com/golang/glog"
@@ -52,6 +53,12 @@ func GatherResults(waitfile string, url string) error {
  s := string(inputFileName)
  glog.Infof("Detected done file, transmitting: (%v)", s)
 
+ // Append a file extension, if there is one
+ filenameParts := strings.SplitN(s, ".", 2)
+ if len(filenameParts) == 2 {
+ url += "." + filenameParts[1]
+ }
+
  // transmit back the results file.
  return DoRequest(url, func() (io.Reader, error) {
  outfile, err := os.Open(s)