fixes according to comments

Author: replay

cmd/mt-whisper-importer-reader/main.go

@@ -269,40 +269,43 @@ func convertWhisperMethod(whisperMethod whisper.AggregationMethod) (schema.Metho
 	}
 }
 
-func getMetric(w *whisper.Whisper, file, name string) (mdata.ArchiveRequest, error) {
-	res := mdata.ArchiveRequest{
-		MetricData: schema.MetricData{
-			Name:  name,
-			Value: 0,
-			Unit:  "unknown",
-			Time:  0,
-			Mtype: "gauge",
-			Tags:  []string{},
-			OrgId: *orgId,
-		},
-	}
-
+func getMetric(w *whisper.Whisper, file, name string) (*mdata.ArchiveRequest, error) {
 	if len(w.Header.Archives) == 0 {
-		return res, fmt.Errorf("Whisper file contains no archives: %q", file)
+		return nil, fmt.Errorf("Whisper file contains no archives: %q", file)
 	}
 
 	method, err := convertWhisperMethod(w.Header.Metadata.AggregationMethod)
 	if err != nil {
-		return res, err
+		return nil, err
 	}
 
-	res.MetricData.Interval = int(w.Header.Archives[0].SecondsPerPoint)
-	res.MetricData.SetId()
-
 	points := make(map[int][]whisper.Point)
 	for i := range w.Header.Archives {
 		p, err := w.DumpArchive(i)
 		if err != nil {
-			return res, fmt.Errorf("Failed to dump archive %d from whisper file %s", i, file)
+			return nil, fmt.Errorf("Failed to dump archive %d from whisper file %s", i, file)
 		}
 		points[i] = p
 	}
 
+	res := &mdata.ArchiveRequest{
+		MetricData: schema.MetricData{
+			Name:     name,
+			Value:    0,
+			Interval: int(w.Header.Archives[0].SecondsPerPoint),
+			Unit:     "unknown",
+			Time:     0,
+			Mtype:    "gauge",
+			Tags:     []string{},
+			OrgId:    *orgId,
+		},
+	}
+	res.MetricData.SetId()
+	mkey, err := schema.MKeyFromString(res.MetricData.Id)
+	if err != nil {
+		panic(err)
+	}
+
 	_, selectedSchema := schemas.Match(res.MetricData.Name, int(w.Header.Archives[0].SecondsPerPoint))
 	conversion := newConversion(w.Header.Archives, points, method)
 	for retIdx, retention := range selectedSchema.Retentions {
@@ -312,11 +315,6 @@ func getMetric(w *whisper.Whisper, file, name string) (mdata.ArchiveRequest, err
 				continue
 			}
 
-			mkey, err := schema.MKeyFromString(res.MetricData.Id)
-			if err != nil {
-				panic(err)
-			}
-
 			encodedChunks := encodedChunksFromPoints(p, uint32(retention.SecondsPerPoint), retention.ChunkSpan)
 			for _, chunk := range encodedChunks {
 				res.ChunkWriteRequests = append(res.ChunkWriteRequests, mdata.NewChunkWriteRequest(

mdata/cwr.go

@@ -7,9 +7,10 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"time"
 
+	"github.com/tinylib/msgp/msgp"
+
 	"github.com/raintank/schema"
 )
 
@@ -42,21 +43,12 @@ type ArchiveRequest struct {
 func (a *ArchiveRequest) MarshalCompressed() (*bytes.Buffer, error) {
 	var buf bytes.Buffer
 
-	// prefix the buffer with version number 1
 	buf.WriteByte(byte(uint8(1)))
 
-	b, err := a.MarshalMsg(nil)
-	if err != nil {
-		return &buf, errors.New(fmt.Sprintf("ERROR: Marshalling metric: %q", err))
-	}
-
 	g := gzip.NewWriter(&buf)
-	_, err = g.Write(b)
-	if err != nil {
-		return &buf, errors.New(fmt.Sprintf("ERROR: Compressing MSGP data: %q", err))
-	}
+	msgp.Encode(g, a)
 
-	err = g.Close()
+	err := g.Close()
 	if err != nil {
 		return &buf, errors.New(fmt.Sprintf("ERROR: Compressing MSGP data: %q", err))
 	}
@@ -65,28 +57,23 @@ func (a *ArchiveRequest) MarshalCompressed() (*bytes.Buffer, error) {
 }
 
 func (a *ArchiveRequest) UnmarshalCompressed(b io.Reader) error {
-	reader := bufio.NewReader(b)
-	versionBuf, err := reader.ReadByte()
-	if err != nil {
-		return errors.New(fmt.Sprintf("ERROR: Failed to read received data: %q", err))
+	versionBuf := make([]byte, 1)
+	readBytes, err := b.Read(versionBuf)
+	if err != nil || readBytes != 1 {
+		return errors.New(fmt.Sprintf("ERROR: Failed to read one byte: %s", err))
 	}
 
-	version := uint8(versionBuf)
+	version := uint8(versionBuf[0])
 	if version != 1 {
 		return errors.New(fmt.Sprintf("ERROR: Only version 1 is supported, received version %d", version))
 	}
 
-	gzipReader, err := gzip.NewReader(reader)
+	gzipReader, err := gzip.NewReader(b)
 	if err != nil {
 		return errors.New(fmt.Sprintf("ERROR: Creating Gzip reader: %q", err))
 	}
 
-	raw, err := ioutil.ReadAll(gzipReader)
-	if err != nil {
-		return errors.New(fmt.Sprintf("ERROR: Decompressing Gzip: %q", err))
-	}
-
-	_, err = a.UnmarshalMsg(raw)
+	err = msgp.Decode(bufio.NewReader(gzipReader), a)
 	if err != nil {
 		return errors.New(fmt.Sprintf("ERROR: Unmarshaling Raw: %q", err))
 	}

mdata/cwr_test.go

@@ -21,13 +21,12 @@ func TestArchiveRequestEncodingDecoding(t *testing.T) {
 	}
 
 	encoded, err := originalRequest.MarshalCompressed()
-	reader := bytes.NewReader(encoded.Bytes())
 	if err != nil {
-		t.Fatalf("Expected no error when encoding request, got %q", err)
+		t.Fatalf("Failed when encoding the request: %s", err)
 	}
 
 	decodedRequest := ArchiveRequest{}
-	err = decodedRequest.UnmarshalCompressed(reader)
+	err = decodedRequest.UnmarshalCompressed(encoded)
 	if err != nil {
 		t.Fatalf("Expected no error when decoding request, got %q", err)
 	}