updated format of cwrs to transfer in importer

Author: replay

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

@@ -309,10 +309,6 @@ func getMetric(w *whisper.Whisper, file, name string) (*mdata.ArchiveRequest, er
 		},
 	}
 	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)
@@ -323,18 +319,10 @@ func getMetric(w *whisper.Whisper, file, name string) (*mdata.ArchiveRequest, er
 				continue
 			}
 
-			var amkey schema.AMKey
-			if retIdx == 0 {
-				amkey = schema.AMKey{MKey: mkey}
-			} else {
-				amkey = schema.GetAMKey(mkey, m, retention.ChunkSpan)
-			}
-
 			encodedChunks := encodedChunksFromPoints(p, uint32(retention.SecondsPerPoint), retention.ChunkSpan)
 			for _, chunk := range encodedChunks {
-				res.ChunkWriteRequests = append(res.ChunkWriteRequests, mdata.NewChunkWriteRequest(
-					nil,
-					amkey,
+				res.ChunkWriteRequests = append(res.ChunkWriteRequests, mdata.NewChunkWriteRequestWithoutOrg(
+					schema.NewArchive(m, retention.ChunkSpan),
 					uint32(retention.MaxRetention()),
 					chunk.Series.T0,
 					chunk.Encode(retention.ChunkSpan),

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

@@ -1,10 +1,12 @@
 package main
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -173,8 +175,15 @@ func (s *Server) healthzHandler(w http.ResponseWriter, req *http.Request) {
 }
 
 func (s *Server) chunksHandler(w http.ResponseWriter, req *http.Request) {
+	orgId, err := getOrgId(req)
+	if err != nil {
+		w.WriteHeader(http.StatusForbidden)
+		w.Write([]byte(err.Error()))
+		return
+	}
+
 	data := mdata.ArchiveRequest{}
-	err := data.UnmarshalCompressed(req.Body)
+	err = data.UnmarshalCompressed(req.Body)
 	if err != nil {
 		throwError(w, fmt.Sprintf("Error decoding cwr stream: %q", err))
 		return
@@ -189,13 +198,15 @@ func (s *Server) chunksHandler(w http.ResponseWriter, req *http.Request) {
 		"Received %d cwrs for metric %s. The first has Key: %s, T0: %d, TTL: %d. The last has Key: %s, T0: %d, TTL: %d",
 		len(data.ChunkWriteRequests),
 		data.MetricData.Name,
-		data.ChunkWriteRequests[0].Key.String(),
+		data.ChunkWriteRequests[0].Archive.String(),
 		data.ChunkWriteRequests[0].T0,
 		data.ChunkWriteRequests[0].TTL,
-		data.ChunkWriteRequests[len(data.ChunkWriteRequests)-1].Key.String(),
+		data.ChunkWriteRequests[len(data.ChunkWriteRequests)-1].Archive.String(),
 		data.ChunkWriteRequests[len(data.ChunkWriteRequests)-1].T0,
 		data.ChunkWriteRequests[len(data.ChunkWriteRequests)-1].TTL)
 
+	data.MetricData.OrgId = orgId
+	data.MetricData.SetId()
 	partition, err := s.partitioner.Partition(&data.MetricData, int32(*numPartitions))
 	if err != nil {
 		throwError(w, fmt.Sprintf("Error partitioning: %q", err))
@@ -210,7 +221,18 @@ func (s *Server) chunksHandler(w http.ResponseWriter, req *http.Request) {
 
 	s.index.AddOrUpdate(mkey, &data.MetricData, partition)
 	for _, cwr := range data.ChunkWriteRequests {
-		cwr := cwr // important because we pass by reference and this var will get overwritten in the next loop
-		s.store.Add(&cwr)
+		cwrWithOrg := cwr.GetChunkWriteRequest(nil, mkey)
+		s.store.Add(&cwrWithOrg)
+	}
+}
+
+func getOrgId(req *http.Request) (int, error) {
+	if orgIdStr := req.Header.Get("X-Org-Id"); len(orgIdStr) > 0 {
+		if orgId, err := strconv.Atoi(orgIdStr); err == nil {
+			return orgId, nil
+		} else {
+			return 0, fmt.Errorf("Invalid value in X-Org-Id header (%s): %s", orgIdStr, err)
+		}
 	}
+	return 0, errors.New("Missing X-Org-Id header")
 }

mdata/cwr.go

@@ -13,30 +13,48 @@ import (
 	"github.com/raintank/schema"
 )
 
+//go:generate msgp
+//msgp:ignore ChunkWriteRequest
+
 type ChunkSaveCallback func()
 
 // ChunkWriteRequest is a request to write a chunk into a store
-//go:generate msgp
 type ChunkWriteRequest struct {
-	Callback  ChunkSaveCallback `msg:"-"`
-	Key       schema.AMKey      `msg:"key,extension"`
+	ChunkWriteRequestPayload
+	Callback ChunkSaveCallback
+	Key      schema.AMKey
+}
+
+func NewChunkWriteRequest(callback ChunkSaveCallback, key schema.AMKey, ttl, t0 uint32, data []byte, ts time.Time) ChunkWriteRequest {
+	return ChunkWriteRequest{ChunkWriteRequestPayload{ttl, t0, data, ts}, callback, key}
+}
+
+// ChunkWriteRequestWithoutOrg is used by the importer utility to send cwrs over the network
+type ChunkWriteRequestWithoutOrg struct {
+	ChunkWriteRequestPayload
+	Archive schema.Archive
+}
+
+func NewChunkWriteRequestWithoutOrg(archive schema.Archive, ttl, t0 uint32, data []byte, ts time.Time) ChunkWriteRequestWithoutOrg {
+	return ChunkWriteRequestWithoutOrg{ChunkWriteRequestPayload{ttl, t0, data, ts}, archive}
+}
+
+func (c *ChunkWriteRequestWithoutOrg) GetChunkWriteRequest(callback ChunkSaveCallback, key schema.MKey) ChunkWriteRequest {
+	return ChunkWriteRequest{c.ChunkWriteRequestPayload, callback, schema.AMKey{MKey: key, Archive: c.Archive}}
+}
+
+type ChunkWriteRequestPayload struct {
 	TTL       uint32
 	T0        uint32
 	Data      []byte
 	Timestamp time.Time
 }
 
-// NewChunkWriteRequest creates a new ChunkWriteRequest
-func NewChunkWriteRequest(callback ChunkSaveCallback, key schema.AMKey, ttl, t0 uint32, data []byte, ts time.Time) ChunkWriteRequest {
-	return ChunkWriteRequest{callback, key, ttl, t0, data, ts}
-}
-
 // ArchiveRequest is a complete representation of a Metric together with some
 // chunk write requests containing data that shall be written into this metric
-//go:generate msgp
 type ArchiveRequest struct {
 	MetricData         schema.MetricData
-	ChunkWriteRequests []ChunkWriteRequest
+	ChunkWriteRequests []ChunkWriteRequestWithoutOrg
 }
 
 func (a *ArchiveRequest) MarshalCompressed() (*bytes.Buffer, error) {