Merge pull request #885 from grafana/orgid-uint

represent orgid in a uint32 to save memory

Author: Dieterbe

api/graphite.go

@@ -54,7 +54,7 @@ type Series struct {
 	Node    cluster.Node
 }
 
-func (s *Server) findSeries(ctx context.Context, orgId int, patterns []string, seenAfter int64) ([]Series, error) {
+func (s *Server) findSeries(ctx context.Context, orgId uint32, patterns []string, seenAfter int64) ([]Series, error) {
 	peers, err := cluster.MembersForQuery()
 	if err != nil {
 		log.Error(3, "HTTP findSeries unable to get peers, %s", err)
@@ -118,7 +118,7 @@ func (s *Server) findSeries(ctx context.Context, orgId int, patterns []string, s
 	return series, nil
 }
 
-func (s *Server) findSeriesLocal(ctx context.Context, orgId int, patterns []string, seenAfter int64) ([]Series, error) {
+func (s *Server) findSeriesLocal(ctx context.Context, orgId uint32, patterns []string, seenAfter int64) ([]Series, error) {
 	result := make([]Series, 0)
 	for _, pattern := range patterns {
 		select {
@@ -149,7 +149,7 @@ func (s *Server) findSeriesLocal(ctx context.Context, orgId int, patterns []stri
 	return result, nil
 }
 
-func (s *Server) findSeriesRemote(ctx context.Context, orgId int, patterns []string, seenAfter int64, peer cluster.Node) ([]Series, error) {
+func (s *Server) findSeriesRemote(ctx context.Context, orgId uint32, patterns []string, seenAfter int64, peer cluster.Node) ([]Series, error) {
 	log.Debug("HTTP Render querying %s/index/find for %d:%q", peer.GetName(), orgId, patterns)
 	data := models.IndexFind{
 		Patterns: patterns,
@@ -370,11 +370,11 @@ func (s *Server) metricsFind(ctx *middleware.Context, request models.GraphiteFin
 	}
 }
 
-func (s *Server) listLocal(orgId int) []idx.Archive {
+func (s *Server) listLocal(orgId uint32) []idx.Archive {
 	return s.MetricIndex.List(orgId)
 }
 
-func (s *Server) listRemote(ctx context.Context, orgId int, peer cluster.Node) ([]idx.Archive, error) {
+func (s *Server) listRemote(ctx context.Context, orgId uint32, peer cluster.Node) ([]idx.Archive, error) {
 	log.Debug("HTTP IndexJson() querying %s/index/list for %d", peer.GetName(), orgId)
 	buf, err := peer.Post(ctx, "listRemote", "/index/list", models.IndexList{OrgId: orgId})
 	if err != nil {
@@ -635,12 +635,12 @@ func (s *Server) metricsDelete(ctx *middleware.Context, req models.MetricsDelete
 	response.Write(ctx, response.NewJson(200, resp, ""))
 }
 
-func (s *Server) metricsDeleteLocal(orgId int, query string) (int, error) {
+func (s *Server) metricsDeleteLocal(orgId uint32, query string) (int, error) {
 	defs, err := s.MetricIndex.Delete(orgId, query)
 	return len(defs), err
 }
 
-func (s *Server) metricsDeleteRemote(ctx context.Context, orgId int, query string, peer cluster.Node) (int, error) {
+func (s *Server) metricsDeleteRemote(ctx context.Context, orgId uint32, query string, peer cluster.Node) (int, error) {
 	log.Debug("HTTP metricDelete calling %s/index/delete for %d:%q", peer.GetName(), orgId, query)
 
 	body := models.IndexDelete{
@@ -673,7 +673,7 @@ func (s *Server) metricsDeleteRemote(ctx context.Context, orgId int, query strin
 // executePlan looks up the needed data, retrieves it, and then invokes the processing
 // note if you do something like sum(foo.*) and all of those metrics happen to be on another node,
 // we will collect all the indidividual series from the peer, and then sum here. that could be optimized
-func (s *Server) executePlan(ctx context.Context, orgId int, plan expr.Plan) ([]models.Series, error) {
+func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan) ([]models.Series, error) {
 
 	minFrom := uint32(math.MaxUint32)
 	var maxTo uint32
@@ -861,7 +861,7 @@ func (s *Server) graphiteTagDetails(ctx *middleware.Context, request models.Grap
 	response.Write(ctx, response.NewJson(200, resp, ""))
 }
 
-func (s *Server) clusterTagDetails(ctx context.Context, orgId int, tag, filter string, from int64) (map[string]uint64, error) {
+func (s *Server) clusterTagDetails(ctx context.Context, orgId uint32, tag, filter string, from int64) (map[string]uint64, error) {
 	result, err := s.MetricIndex.TagDetails(orgId, tag, filter, from)
 	if err != nil {
 		return nil, err
@@ -923,7 +923,7 @@ func (s *Server) graphiteTagFindSeries(ctx *middleware.Context, request models.G
 	response.Write(ctx, response.NewJson(200, seriesNames, ""))
 }
 
-func (s *Server) clusterFindByTag(ctx context.Context, orgId int, expressions []string, from int64) ([]Series, error) {
+func (s *Server) clusterFindByTag(ctx context.Context, orgId uint32, expressions []string, from int64) ([]Series, error) {
 	seriesSet := make(map[string]Series)
 
 	result, err := s.MetricIndex.FindByTag(orgId, expressions, from)
@@ -1005,7 +1005,7 @@ func (s *Server) graphiteTags(ctx *middleware.Context, request models.GraphiteTa
 	response.Write(ctx, response.NewJson(200, resp, ""))
 }
 
-func (s *Server) clusterTags(ctx context.Context, orgId int, filter string, from int64) ([]string, error) {
+func (s *Server) clusterTags(ctx context.Context, orgId uint32, filter string, from int64) ([]string, error) {
 	result, err := s.MetricIndex.Tags(orgId, filter, from)
 	if err != nil {
 		return nil, err
@@ -1068,7 +1068,7 @@ func (s *Server) graphiteAutoCompleteTags(ctx *middleware.Context, request model
 	response.Write(ctx, response.NewJson(200, tags, ""))
 }
 
-func (s *Server) clusterAutoCompleteTags(ctx context.Context, orgId int, prefix string, expressions []string, from int64, limit uint) ([]string, error) {
+func (s *Server) clusterAutoCompleteTags(ctx context.Context, orgId uint32, prefix string, expressions []string, from int64, limit uint) ([]string, error) {
 	result, err := s.MetricIndex.FindTags(orgId, prefix, expressions, from, limit)
 	if err != nil {
 		return nil, err
@@ -1123,7 +1123,7 @@ func (s *Server) graphiteAutoCompleteTagValues(ctx *middleware.Context, request
 	response.Write(ctx, response.NewJson(200, resp, ""))
 }
 
-func (s *Server) clusterAutoCompleteTagValues(ctx context.Context, orgId int, tag, prefix string, expressions []string, from int64, limit uint) ([]string, error) {
+func (s *Server) clusterAutoCompleteTagValues(ctx context.Context, orgId uint32, tag, prefix string, expressions []string, from int64, limit uint) ([]string, error) {
 	result, err := s.MetricIndex.FindTagValues(orgId, tag, prefix, expressions, from, limit)
 	if err != nil {
 		return nil, err

api/middleware/middleware.go

@@ -12,7 +12,7 @@ import (
 
 type Context struct {
 	*macaron.Context
-	OrgId int
+	OrgId uint32
 	Body  io.ReadCloser
 }
 
@@ -31,7 +31,7 @@ func OrgMiddleware(multiTenant bool) macaron.Handler {
 	}
 }
 
-func getOrg(req *http.Request, multiTenant bool) (int, error) {
+func getOrg(req *http.Request, multiTenant bool) (uint32, error) {
 	if !multiTenant {
 		return 1, nil
 	}
@@ -40,10 +40,10 @@ func getOrg(req *http.Request, multiTenant bool) (int, error) {
 		return 0, nil
 	}
 	org, err := strconv.Atoi(orgStr)
-	if err != nil {
+	if err != nil || org < 1 {
 		return 0, errors.New("bad org-id")
 	}
-	return org, nil
+	return uint32(org), nil
 }
 
 func RequireOrg() macaron.Handler {

api/models/ccache.go

@@ -9,7 +9,7 @@ type CCacheDelete struct {
 	Patterns []string `json:"patterns" form:"patterns" `
 	// tag expressions to select series
 	Expr      []string `json:"expr" form:"expr"`
-	OrgId     int      `json:"orgId" form:"orgId" binding:"Required"`
+	OrgId     uint32   `json:"orgId" form:"orgId" binding:"Required"`
 	Propagate bool     `json:"propagate" form:"propagate"`
 }
 

api/models/node.go

@@ -26,7 +26,7 @@ type ClusterMembersResp struct {
 }
 
 type IndexList struct {
-	OrgId int `json:"orgId" form:"orgId" binding:"Required"`
+	OrgId uint32 `json:"orgId" form:"orgId" binding:"Required"`
 }
 
 func (i IndexList) Trace(span opentracing.Span) {
@@ -37,7 +37,7 @@ func (i IndexList) TraceDebug(span opentracing.Span) {
 }
 
 type IndexFindByTag struct {
-	OrgId int      `json:"orgId" binding:"Required"`
+	OrgId uint32   `json:"orgId" binding:"Required"`
 	Expr  []string `json:"expressions"`
 	From  int64    `json:"from"`
 }
@@ -52,7 +52,7 @@ func (i IndexFindByTag) TraceDebug(span opentracing.Span) {
 }
 
 type IndexTagDetails struct {
-	OrgId  int    `json:"orgId" binding:"Required"`
+	OrgId  uint32 `json:"orgId" binding:"Required"`
 	Filter string `json:"filter"`
 	Tag    string `json:"tag" binding:"Required"`
 	From   int64  `json:"from"`
@@ -69,7 +69,7 @@ func (i IndexTagDetails) TraceDebug(span opentracing.Span) {
 }
 
 type IndexTags struct {
-	OrgId  int    `json:"orgId" binding:"Required"`
+	OrgId  uint32 `json:"orgId" binding:"Required"`
 	Filter string `json:"filter"`
 	From   int64  `json:"from"`
 }
@@ -84,7 +84,7 @@ func (i IndexTags) TraceDebug(span opentracing.Span) {
 }
 
 type IndexAutoCompleteTags struct {
-	OrgId  int      `json:"orgId" binding:"Required"`
+	OrgId  uint32   `json:"orgId" binding:"Required"`
 	Prefix string   `json:"Prefix"`
 	Expr   []string `json:"expressions"`
 	From   int64    `json:"from"`
@@ -103,7 +103,7 @@ func (i IndexAutoCompleteTags) TraceDebug(span opentracing.Span) {
 }
 
 type IndexAutoCompleteTagValues struct {
-	OrgId  int      `json:"orgId" binding:"Required"`
+	OrgId  uint32   `json:"orgId" binding:"Required"`
 	Tag    string   `json:"tag"`
 	Prefix string   `json:"prefix"`
 	Expr   []string `json:"expressions"`
@@ -124,7 +124,7 @@ func (i IndexAutoCompleteTagValues) TraceDebug(span opentracing.Span) {
 }
 
 type IndexTagDelSeries struct {
-	OrgId int      `json:"orgId" binding:"Required"`
+	OrgId uint32   `json:"orgId" binding:"Required"`
 	Paths []string `json:"path" form:"path"`
 }
 
@@ -142,7 +142,7 @@ type IndexGet struct {
 
 type IndexFind struct {
 	Patterns []string `json:"patterns" form:"patterns" binding:"Required"`
-	OrgId    int      `json:"orgId" form:"orgId" binding:"Required"`
+	OrgId    uint32   `json:"orgId" form:"orgId" binding:"Required"`
 	From     int64    `json:"from" form:"from"`
 }
 
@@ -177,7 +177,7 @@ func (g GetData) TraceDebug(span opentracing.Span) {
 
 type IndexDelete struct {
 	Query string `json:"query" form:"query" binding:"Required"`
-	OrgId int    `json:"orgId" form:"orgId" binding:"Required"`
+	OrgId uint32 `json:"orgId" form:"orgId" binding:"Required"`
 }
 
 func (i IndexDelete) Trace(span opentracing.Span) {

api/prometheus_querier.go

@@ -21,20 +21,20 @@ import (
 func (s *Server) Querier(ctx context.Context, min, max int64) (storage.Querier, error) {
 	from := uint32(min / 1000)
 	to := uint32(max / 1000)
-	return NewQuerier(ctx, s, from, to, ctx.Value(orgID("org-id")).(int), false), nil
+	return NewQuerier(ctx, s, from, to, ctx.Value(orgID("org-id")).(uint32), false), nil
 }
 
 // querier implements Prometheus' Querier interface
 type querier struct {
 	*Server
 	from         uint32
 	to           uint32
-	OrgID        int
+	OrgID        uint32
 	metadataOnly bool
 	ctx          context.Context
 }
 
-func NewQuerier(ctx context.Context, s *Server, from, to uint32, orgId int, metadataOnly bool) storage.Querier {
+func NewQuerier(ctx context.Context, s *Server, from, to uint32, orgId uint32, metadataOnly bool) storage.Querier {
 	return &querier{
 		s,
 		from,

cmd/mt-index-migrate/main.go

@@ -156,7 +156,7 @@ func getDefs(session *gocql.Session, defsChan chan *schema.MetricDefinition) {
 		}
 		mdef := schema.MetricDefinition{
 			Id:         mkey,
-			OrgId:      orgId,
+			OrgId:      uint32(orgId),
 			Partition:  partition,
 			Name:       name,
 			Interval:   interval,

idx/cassandra/cassandra.go

@@ -399,10 +399,13 @@ func (c *CasIdx) load(defs []schema.MetricDefinition, iter cqlIterator, cutoff u
 			log.Error(3, "cassandra-idx: load() could not parse ID %q: %s -> skipping", id, err)
 			continue
 		}
+		if orgId < 0 {
+			orgId = int(idx.OrgIdPublic)
+		}
 
 		mdef := &schema.MetricDefinition{
 			Id:         mkey,
-			OrgId:      orgId,
+			OrgId:      uint32(orgId),
 			Partition:  partition,
 			Name:       name,
 			Interval:   interval,
@@ -487,7 +490,7 @@ func (c *CasIdx) processWriteQueue() {
 	c.wg.Done()
 }
 
-func (c *CasIdx) Delete(orgId int, pattern string) ([]idx.Archive, error) {
+func (c *CasIdx) Delete(orgId uint32, pattern string) ([]idx.Archive, error) {
 	pre := time.Now()
 	defs, err := c.MemoryIdx.Delete(orgId, pattern)
 	if err != nil {

idx/cassandra/cassandra_test.go

@@ -104,14 +104,14 @@ func getRandomString(n int, alphabets ...byte) string {
 	return string(bytes)
 }
 
-func getMetricData(orgId, depth, count, interval int, prefix string) []*schema.MetricData {
+func getMetricData(orgId uint32, depth, count, interval int, prefix string) []*schema.MetricData {
 	data := make([]*schema.MetricData, count)
 	series := getSeriesNames(depth, count, prefix)
 	for i, s := range series {
 
 		data[i] = &schema.MetricData{
 			Name:     s,
-			OrgId:    orgId,
+			OrgId:    int(orgId),
 			Interval: interval,
 		}
 		data[i].SetId()
@@ -131,7 +131,7 @@ func TestGetAddKey(t *testing.T) {
 	org2Series := getMetricData(2, 2, 5, 10, "metric.org2")
 
 	for _, series := range [][]*schema.MetricData{publicSeries, org1Series, org2Series} {
-		orgId := series[0].OrgId
+		orgId := uint32(series[0].OrgId)
 		Convey(fmt.Sprintf("When indexing metrics for orgId %d", orgId), t, func() {
 			for _, s := range series {
 				mkey, err := schema.MKeyFromString(s.Id)

idx/idx.go

@@ -9,7 +9,7 @@ import (
 	schema "gopkg.in/raintank/schema.v1"
 )
 
-var OrgIdPublic = 0
+var OrgIdPublic = uint32(0)
 
 var (
 	BothBranchAndLeaf = errors.New("node can't be both branch and leaf")
@@ -69,23 +69,23 @@ type MetricIndex interface {
 	Get(key schema.MKey) (Archive, bool)
 
 	// GetPath returns the archives under the given path.
-	GetPath(orgId int, path string) []Archive
+	GetPath(orgId uint32, path string) []Archive
 
 	// Delete deletes items from the index
 	// If the pattern matches a branch node, then
 	// all leaf nodes on that branch are deleted. So if the pattern is
 	// "*", all items in the index are deleted.
 	// It returns a copy of all of the Archives deleted.
-	Delete(orgId int, pattern string) ([]Archive, error)
+	Delete(orgId uint32, pattern string) ([]Archive, error)
 
 	// Find searches the index for matching nodes.
 	// * orgId describes the org to search in (public data in orgIdPublic is automatically included)
 	// * pattern is handled like graphite does. see https://graphite.readthedocs.io/en/latest/render_api.html#paths-and-wildcards
 	// * from is a unix timestamp. series not updated since then are excluded.
-	Find(orgId int, pattern string, from int64) ([]Node, error)
+	Find(orgId uint32, pattern string, from int64) ([]Node, error)
 
 	// List returns all Archives for the passed OrgId and the public orgId
-	List(orgId int) []Archive
+	List(orgId uint32) []Archive
 
 	// Prune deletes all metrics that haven't been seen since the given timestamp.
 	// It returns all Archives deleted and any error encountered.
@@ -99,24 +99,24 @@ type MetricIndex interface {
 	// where the LastUpdate time is >= from will be returned as results.
 	// The returned results are not deduplicated and in certain cases it is possible
 	// that duplicate entries will be returned.
-	FindByTag(orgId int, expressions []string, from int64) ([]Node, error)
+	FindByTag(orgId uint32, expressions []string, from int64) ([]Node, error)
 
 	// Tags returns a list of all tag keys associated with the metrics of a given
 	// organization. The return values are filtered by the regex in the second parameter.
 	// If the third parameter is >0 then only metrics will be accounted of which the
 	// LastUpdate time is >= the given value.
-	Tags(orgId int, filter string, from int64) ([]string, error)
+	Tags(orgId uint32, filter string, from int64) ([]string, error)
 
 	// FindTags generates a list of possible tags that could complete a
 	// given prefix. It also accepts additional tag conditions to further narrow
 	// down the result set in the format of graphite's tag queries
-	FindTags(orgId int, prefix string, expressions []string, from int64, limit uint) ([]string, error)
+	FindTags(orgId uint32, prefix string, expressions []string, from int64, limit uint) ([]string, error)
 
 	// FindTagValues generates a list of possible values that could
 	// complete a given value prefix. It requires a tag to be specified and only values
 	// of the given tag will be returned. It also accepts additional conditions to
 	// further narrow down the result set in the format of graphite's tag queries
-	FindTagValues(orgId int, tag string, prefix string, expressions []string, from int64, limit uint) ([]string, error)
+	FindTagValues(orgId uint32, tag string, prefix string, expressions []string, from int64, limit uint) ([]string, error)
 
 	// TagDetails returns a list of all values associated with a given tag key in the
 	// given org. The occurrences of each value is counted and the count is referred to by
@@ -125,9 +125,9 @@ type MetricIndex interface {
 	// the values before accounting for them.
 	// If the fourth parameter is > 0 then only those metrics of which the LastUpdate
 	// time is >= the from timestamp will be included.
-	TagDetails(orgId int, key string, filter string, from int64) (map[string]uint64, error)
+	TagDetails(orgId uint32, key string, filter string, from int64) (map[string]uint64, error)
 
 	// DeleteTagged deletes the specified series from the tag index and also the
 	// DefById index.
-	DeleteTagged(orgId int, paths []string) ([]Archive, error)
+	DeleteTagged(orgId uint32, paths []string) ([]Archive, error)
 }